"""create knowledge module Revision ID: b6f3d8c1a920 Revises: a5e4b7c9d012 Create Date: 2026-07-03 12:00:00.000000 """ from typing import Sequence, Union from alembic import op import sqlalchemy as sa revision: str = "b6f3d8c1a920" down_revision: Union[str, Sequence[str], None] = "a5e4b7c9d012" branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None KNOWLEDGE_PERMISSIONS = [ ("knowledge.read", "Wissen lesen", "Wissensdatenbank anzeigen", "knowledge"), ("knowledge.create", "Wissen erstellen", "Wissensdatenbank-Einträge erstellen", "knowledge"), ("knowledge.update", "Wissen bearbeiten", "Wissensdatenbank-Einträge aktualisieren", "knowledge"), ("knowledge.delete", "Wissen löschen", "Wissensdatenbank-Einträge entfernen", "knowledge"), ("knowledge.upload", "Wissen hochladen", "Wissensdokumente hochladen", "knowledge"), ("knowledge.download", "Wissen herunterladen", "Wissensdokumente herunterladen", "knowledge"), ] ROLE_PERMISSIONS = { "administrator": [permission[0] for permission in KNOWLEDGE_PERMISSIONS], "management": ["knowledge.read", "knowledge.download"], "technician": ["knowledge.read", "knowledge.create", "knowledge.update", "knowledge.upload", "knowledge.download"], "support": ["knowledge.read", "knowledge.download"], } def insert_permission(name: str, display_name: str, description: str, module: str) -> None: op.execute( sa.text( """ INSERT INTO permissions (name, display_name, description, module) SELECT :name, :display_name, :description, :module WHERE NOT EXISTS (SELECT 1 FROM permissions WHERE name = :name) """ ).bindparams(name=name, display_name=display_name, description=description, module=module) ) def grant_permission(role_name: str, permission_name: str) -> None: op.execute( sa.text( """ INSERT INTO role_permissions (role_id, permission_id) SELECT roles.id, permissions.id FROM roles JOIN permissions ON permissions.name = :permission_name WHERE roles.name = :role_name AND NOT EXISTS ( SELECT 1 FROM role_permissions WHERE role_permissions.role_id = roles.id AND role_permissions.permission_id = permissions.id ) """ ).bindparams(role_name=role_name, permission_name=permission_name) ) def upgrade() -> None: op.create_table( "knowledge_manufacturers", sa.Column("id", sa.Integer(), nullable=False), sa.Column("name", sa.String(length=160), nullable=False), sa.Column("slug", sa.String(length=180), nullable=False), sa.Column("website", sa.String(length=255), server_default="", nullable=False), sa.Column("notes", sa.Text(), server_default="", nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.PrimaryKeyConstraint("id"), ) op.create_index(op.f("ix_knowledge_manufacturers_name"), "knowledge_manufacturers", ["name"], unique=True) op.create_index(op.f("ix_knowledge_manufacturers_slug"), "knowledge_manufacturers", ["slug"], unique=True) op.create_table( "knowledge_categories", sa.Column("id", sa.Integer(), nullable=False), sa.Column("name", sa.String(length=120), nullable=False), sa.Column("slug", sa.String(length=140), nullable=False), sa.Column("description", sa.Text(), server_default="", nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.PrimaryKeyConstraint("id"), ) op.create_index(op.f("ix_knowledge_categories_name"), "knowledge_categories", ["name"], unique=True) op.create_index(op.f("ix_knowledge_categories_slug"), "knowledge_categories", ["slug"], unique=True) op.create_table( "knowledge_devices", sa.Column("id", sa.Integer(), nullable=False), sa.Column("manufacturer_id", sa.Integer(), nullable=False), sa.Column("name", sa.String(length=180), nullable=False), sa.Column("slug", sa.String(length=200), nullable=False), sa.Column("model_number", sa.String(length=120), server_default="", nullable=False), sa.Column("device_type", sa.String(length=80), server_default="", nullable=False), sa.Column("frequency_range", sa.String(length=120), server_default="", nullable=False), sa.Column("production_year_from", sa.Integer(), nullable=True), sa.Column("production_year_to", sa.Integer(), nullable=True), sa.Column("notes", sa.Text(), server_default="", nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.ForeignKeyConstraint(["manufacturer_id"], ["knowledge_manufacturers.id"], ondelete="CASCADE"), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("manufacturer_id", "slug", name="uq_knowledge_devices_manufacturer_slug"), ) op.create_index(op.f("ix_knowledge_devices_manufacturer_id"), "knowledge_devices", ["manufacturer_id"], unique=False) op.create_index(op.f("ix_knowledge_devices_name"), "knowledge_devices", ["name"], unique=False) op.create_index(op.f("ix_knowledge_devices_slug"), "knowledge_devices", ["slug"], unique=False) op.create_index(op.f("ix_knowledge_devices_model_number"), "knowledge_devices", ["model_number"], unique=False) op.create_index(op.f("ix_knowledge_devices_device_type"), "knowledge_devices", ["device_type"], unique=False) op.create_table( "knowledge_documents", sa.Column("id", sa.Integer(), nullable=False), sa.Column("manufacturer_id", sa.Integer(), nullable=False), sa.Column("device_id", sa.Integer(), nullable=True), sa.Column("title", sa.String(length=255), nullable=False), sa.Column("slug", sa.String(length=280), nullable=False), sa.Column("document_type", sa.String(length=80), nullable=False), sa.Column("language", sa.String(length=20), server_default="de", nullable=False), sa.Column("file_name", sa.String(length=255), server_default="", nullable=False), sa.Column("file_path", sa.String(length=500), server_default="", nullable=False), sa.Column("mime_type", sa.String(length=120), server_default="", nullable=False), sa.Column("file_size", sa.Integer(), server_default="0", nullable=False), sa.Column("checksum_sha256", sa.String(length=64), server_default="", nullable=False), sa.Column("external_url", sa.String(length=500), server_default="", nullable=False), sa.Column("paperless_document_id", sa.String(length=120), server_default="", nullable=False), sa.Column("description", sa.Text(), server_default="", nullable=False), sa.Column("tags", sa.JSON(), server_default="[]", nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.ForeignKeyConstraint(["device_id"], ["knowledge_devices.id"], ondelete="SET NULL"), sa.ForeignKeyConstraint(["manufacturer_id"], ["knowledge_manufacturers.id"], ondelete="RESTRICT"), sa.PrimaryKeyConstraint("id"), ) op.create_index(op.f("ix_knowledge_documents_manufacturer_id"), "knowledge_documents", ["manufacturer_id"], unique=False) op.create_index(op.f("ix_knowledge_documents_device_id"), "knowledge_documents", ["device_id"], unique=False) op.create_index(op.f("ix_knowledge_documents_title"), "knowledge_documents", ["title"], unique=False) op.create_index(op.f("ix_knowledge_documents_slug"), "knowledge_documents", ["slug"], unique=True) op.create_index(op.f("ix_knowledge_documents_document_type"), "knowledge_documents", ["document_type"], unique=False) op.create_index(op.f("ix_knowledge_documents_language"), "knowledge_documents", ["language"], unique=False) op.create_index(op.f("ix_knowledge_documents_checksum_sha256"), "knowledge_documents", ["checksum_sha256"], unique=False) op.create_index(op.f("ix_knowledge_documents_paperless_document_id"), "knowledge_documents", ["paperless_document_id"], unique=False) op.create_table( "knowledge_notes", sa.Column("id", sa.Integer(), nullable=False), sa.Column("manufacturer_id", sa.Integer(), nullable=True), sa.Column("device_id", sa.Integer(), nullable=True), sa.Column("title", sa.String(length=255), nullable=False), sa.Column("content", sa.Text(), nullable=False), sa.Column("note_type", sa.String(length=80), nullable=False), sa.Column("severity", sa.String(length=40), server_default="", nullable=False), sa.Column("tags", sa.JSON(), server_default="[]", nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.ForeignKeyConstraint(["device_id"], ["knowledge_devices.id"], ondelete="SET NULL"), sa.ForeignKeyConstraint(["manufacturer_id"], ["knowledge_manufacturers.id"], ondelete="SET NULL"), sa.PrimaryKeyConstraint("id"), ) op.create_index(op.f("ix_knowledge_notes_manufacturer_id"), "knowledge_notes", ["manufacturer_id"], unique=False) op.create_index(op.f("ix_knowledge_notes_device_id"), "knowledge_notes", ["device_id"], unique=False) op.create_index(op.f("ix_knowledge_notes_title"), "knowledge_notes", ["title"], unique=False) op.create_index(op.f("ix_knowledge_notes_note_type"), "knowledge_notes", ["note_type"], unique=False) op.create_index(op.f("ix_knowledge_notes_severity"), "knowledge_notes", ["severity"], unique=False) for permission in KNOWLEDGE_PERMISSIONS: insert_permission(*permission) for role_name, permission_names in ROLE_PERMISSIONS.items(): for permission_name in permission_names: grant_permission(role_name, permission_name) def downgrade() -> None: op.execute( """ DELETE FROM role_permissions USING permissions WHERE role_permissions.permission_id = permissions.id AND permissions.module = 'knowledge' """ ) op.execute("DELETE FROM permissions WHERE module = 'knowledge'") op.drop_index(op.f("ix_knowledge_notes_severity"), table_name="knowledge_notes") op.drop_index(op.f("ix_knowledge_notes_note_type"), table_name="knowledge_notes") op.drop_index(op.f("ix_knowledge_notes_title"), table_name="knowledge_notes") op.drop_index(op.f("ix_knowledge_notes_device_id"), table_name="knowledge_notes") op.drop_index(op.f("ix_knowledge_notes_manufacturer_id"), table_name="knowledge_notes") op.drop_table("knowledge_notes") op.drop_index(op.f("ix_knowledge_documents_paperless_document_id"), table_name="knowledge_documents") op.drop_index(op.f("ix_knowledge_documents_checksum_sha256"), table_name="knowledge_documents") op.drop_index(op.f("ix_knowledge_documents_language"), table_name="knowledge_documents") op.drop_index(op.f("ix_knowledge_documents_document_type"), table_name="knowledge_documents") op.drop_index(op.f("ix_knowledge_documents_slug"), table_name="knowledge_documents") op.drop_index(op.f("ix_knowledge_documents_title"), table_name="knowledge_documents") op.drop_index(op.f("ix_knowledge_documents_device_id"), table_name="knowledge_documents") op.drop_index(op.f("ix_knowledge_documents_manufacturer_id"), table_name="knowledge_documents") op.drop_table("knowledge_documents") op.drop_index(op.f("ix_knowledge_devices_device_type"), table_name="knowledge_devices") op.drop_index(op.f("ix_knowledge_devices_model_number"), table_name="knowledge_devices") op.drop_index(op.f("ix_knowledge_devices_slug"), table_name="knowledge_devices") op.drop_index(op.f("ix_knowledge_devices_name"), table_name="knowledge_devices") op.drop_index(op.f("ix_knowledge_devices_manufacturer_id"), table_name="knowledge_devices") op.drop_table("knowledge_devices") op.drop_index(op.f("ix_knowledge_categories_slug"), table_name="knowledge_categories") op.drop_index(op.f("ix_knowledge_categories_name"), table_name="knowledge_categories") op.drop_table("knowledge_categories") op.drop_index(op.f("ix_knowledge_manufacturers_slug"), table_name="knowledge_manufacturers") op.drop_index(op.f("ix_knowledge_manufacturers_name"), table_name="knowledge_manufacturers") op.drop_table("knowledge_manufacturers")