96 lines
4.5 KiB
Python
96 lines
4.5 KiB
Python
|
"""Fondations centre documentaire et planning interne des salles."""
|
||
|
|
from alembic import op
|
||
|
|
import sqlalchemy as sa
|
||
|
|
|
||
|
|
|
||
|
|
revision = "g1a2b3c4d5e6"
|
||
|
|
down_revision = "9f4a1b2c3d5e"
|
||
|
|
branch_labels = None
|
||
|
|
depends_on = None
|
||
|
|
|
||
|
|
|
||
|
|
def _column_names(table):
|
||
|
|
bind = op.get_bind()
|
||
|
|
inspector = sa.inspect(bind)
|
||
|
|
return {column["name"] for column in inspector.get_columns(table)}
|
||
|
|
|
||
|
|
|
||
|
|
def upgrade():
|
||
|
|
if "document_type" not in _column_names("intervention_documents"):
|
||
|
|
op.add_column(
|
||
|
|
"intervention_documents",
|
||
|
|
sa.Column("document_type", sa.String(30), nullable=False, server_default="autre"),
|
||
|
|
)
|
||
|
|
op.create_index("ix_intervention_documents_document_type", "intervention_documents", ["document_type"])
|
||
|
|
if "document_type" not in _column_names("equipment_documents"):
|
||
|
|
op.add_column(
|
||
|
|
"equipment_documents",
|
||
|
|
sa.Column("document_type", sa.String(30), nullable=False, server_default="autre"),
|
||
|
|
)
|
||
|
|
op.create_index("ix_equipment_documents_document_type", "equipment_documents", ["document_type"])
|
||
|
|
|
||
|
|
room_columns = _column_names("room_schedules")
|
||
|
|
additions = (
|
||
|
|
("source", sa.String(20), "manual"),
|
||
|
|
("external_id", sa.String(191), None),
|
||
|
|
("last_synced_at", sa.DateTime(), None),
|
||
|
|
("source_updated_at", sa.DateTime(), None),
|
||
|
|
("protected_from_sync", sa.Boolean(), False),
|
||
|
|
("resolution_status", sa.String(20), "active"),
|
||
|
|
("valid_from", sa.Date(), None),
|
||
|
|
("valid_to", sa.Date(), None),
|
||
|
|
("event_type", sa.String(30), "cours"),
|
||
|
|
("conflict_note", sa.Text(), None),
|
||
|
|
)
|
||
|
|
for name, column_type, default in additions:
|
||
|
|
if name not in room_columns:
|
||
|
|
kwargs = {"nullable": False} if name in {"source", "protected_from_sync", "resolution_status", "event_type"} else {"nullable": True}
|
||
|
|
if default is not None:
|
||
|
|
kwargs["server_default"] = sa.text("1" if default is True else "0" if default is False else repr(default))
|
||
|
|
op.add_column("room_schedules", sa.Column(name, column_type, **kwargs))
|
||
|
|
for name in ("source", "external_id", "protected_from_sync", "resolution_status"):
|
||
|
|
index_name = f"ix_room_schedules_{name}"
|
||
|
|
if index_name not in {i["name"] for i in sa.inspect(op.get_bind()).get_indexes("room_schedules")}:
|
||
|
|
op.create_index(index_name, "room_schedules", [name])
|
||
|
|
|
||
|
|
bind = op.get_bind()
|
||
|
|
bind.execute(sa.text(
|
||
|
|
"""INSERT INTO permissions (code, name, module, action, description, is_active)
|
||
|
|
SELECT 'documents.view', 'Consulter le centre documentaire', 'documents', 'view',
|
||
|
|
'Permet de rechercher et consulter les documents autorisés des interventions, équipements et produits.', 1
|
||
|
|
WHERE NOT EXISTS (SELECT 1 FROM permissions WHERE code='documents.view')"""
|
||
|
|
))
|
||
|
|
bind.execute(sa.text(
|
||
|
|
"""INSERT INTO role_permissions (role_id, permission_id, effect)
|
||
|
|
SELECT r.id, p.id, 'allow'
|
||
|
|
FROM roles r CROSS JOIN permissions p
|
||
|
|
WHERE p.code='documents.view'
|
||
|
|
AND r.slug IN ('admin','responsable_gmao','technicien','assistant_prevention','demandeur','lecture')
|
||
|
|
AND NOT EXISTS (
|
||
|
|
SELECT 1 FROM role_permissions rp
|
||
|
|
WHERE rp.role_id=r.id AND rp.permission_id=p.id
|
||
|
|
)"""
|
||
|
|
))
|
||
|
|
|
||
|
|
|
||
|
|
def downgrade():
|
||
|
|
bind = op.get_bind()
|
||
|
|
bind.execute(sa.text(
|
||
|
|
"""DELETE rp FROM role_permissions rp
|
||
|
|
JOIN permissions p ON p.id=rp.permission_id
|
||
|
|
WHERE p.code='documents.view'"""
|
||
|
|
))
|
||
|
|
bind.execute(sa.text("DELETE FROM permissions WHERE code='documents.view'"))
|
||
|
|
for name in ("source", "external_id", "last_synced_at", "source_updated_at", "protected_from_sync", "resolution_status", "valid_from", "valid_to", "event_type", "conflict_note"):
|
||
|
|
if name in _column_names("room_schedules"):
|
||
|
|
index_name = f"ix_room_schedules_{name}"
|
||
|
|
if index_name in {i["name"] for i in sa.inspect(bind).get_indexes("room_schedules")}:
|
||
|
|
op.drop_index(index_name, table_name="room_schedules")
|
||
|
|
op.drop_column("room_schedules", name)
|
||
|
|
if "document_type" in _column_names("equipment_documents"):
|
||
|
|
op.drop_index("ix_equipment_documents_document_type", table_name="equipment_documents")
|
||
|
|
op.drop_column("equipment_documents", "document_type")
|
||
|
|
if "document_type" in _column_names("intervention_documents"):
|
||
|
|
op.drop_index("ix_intervention_documents_document_type", table_name="intervention_documents")
|
||
|
|
op.drop_column("intervention_documents", "document_type")
|