"""create repair estimates Revision ID: b4e8f1c7a205 Revises: a1c5f9e2d430 Create Date: 2026-07-05 10:00:00.000000 """ from typing import Sequence, Union from alembic import op import sqlalchemy as sa revision: str = "b4e8f1c7a205" down_revision: Union[str, Sequence[str], None] = "a1c5f9e2d430" branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None ESTIMATE_PERMISSIONS = [ ("repair_estimates.read", "Kostenvoranschläge lesen", "Kostenvoranschläge anzeigen", "repair_estimates"), ("repair_estimates.create", "Kostenvoranschläge erstellen", "Kostenvoranschläge anlegen", "repair_estimates"), ("repair_estimates.update", "Kostenvoranschläge bearbeiten", "Kostenvoranschläge aktualisieren", "repair_estimates"), ("repair_estimates.delete", "Kostenvoranschläge löschen", "Kostenvoranschläge entfernen", "repair_estimates"), ("repair_estimates.send", "Kostenvoranschläge senden", "Kostenvoranschläge an Kunden senden", "repair_estimates"), ] def upgrade() -> None: op.create_table( "repair_estimates", sa.Column("id", sa.Integer(), nullable=False), sa.Column("repair_id", sa.Integer(), nullable=False), sa.Column("estimate_number", sa.String(length=20), nullable=False), sa.Column("status", sa.String(length=40), server_default="draft", nullable=False), sa.Column("title", sa.String(length=255), nullable=False), sa.Column("customer_message", sa.Text(), server_default="", nullable=False), sa.Column("internal_note", sa.Text(), nullable=True), sa.Column("subtotal_cents", sa.Integer(), server_default="0", nullable=False), sa.Column("tax_rate_percent", sa.Numeric(5, 2), server_default="19.00", nullable=False), sa.Column("tax_cents", sa.Integer(), server_default="0", nullable=False), sa.Column("total_cents", sa.Integer(), server_default="0", nullable=False), sa.Column("currency", sa.String(length=3), server_default="EUR", nullable=False), sa.Column("valid_until", sa.Date(), nullable=True), sa.Column("sent_at", sa.DateTime(timezone=True), nullable=True), sa.Column("approved_at", sa.DateTime(timezone=True), nullable=True), sa.Column("declined_at", sa.DateTime(timezone=True), nullable=True), sa.Column("customer_response_message", sa.Text(), nullable=True), sa.Column("created_by_user_id", sa.Integer(), nullable=True), 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(["created_by_user_id"], ["users.id"], ondelete="SET NULL"), sa.ForeignKeyConstraint(["repair_id"], ["repairs.id"], ondelete="CASCADE"), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("estimate_number"), ) op.create_index(op.f("ix_repair_estimates_created_by_user_id"), "repair_estimates", ["created_by_user_id"], unique=False) op.create_index(op.f("ix_repair_estimates_estimate_number"), "repair_estimates", ["estimate_number"], unique=True) op.create_index(op.f("ix_repair_estimates_repair_id"), "repair_estimates", ["repair_id"], unique=False) op.create_index(op.f("ix_repair_estimates_status"), "repair_estimates", ["status"], unique=False) op.create_table( "repair_estimate_items", sa.Column("id", sa.Integer(), nullable=False), sa.Column("estimate_id", sa.Integer(), nullable=False), sa.Column("position", sa.Integer(), nullable=False), sa.Column("item_type", sa.String(length=40), nullable=False), sa.Column("title", sa.String(length=255), nullable=False), sa.Column("description", sa.Text(), nullable=True), sa.Column("quantity", sa.Numeric(10, 2), server_default="1.00", nullable=False), sa.Column("unit", sa.String(length=40), server_default="Stk.", nullable=False), sa.Column("unit_price_cents", sa.Integer(), server_default="0", nullable=False), sa.Column("total_cents", sa.Integer(), server_default="0", 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(["estimate_id"], ["repair_estimates.id"], ondelete="CASCADE"), sa.PrimaryKeyConstraint("id"), ) op.create_index(op.f("ix_repair_estimate_items_estimate_id"), "repair_estimate_items", ["estimate_id"], unique=False) op.create_index(op.f("ix_repair_estimate_items_item_type"), "repair_estimate_items", ["item_type"], unique=False) op.create_table( "repair_estimate_events", sa.Column("id", sa.Integer(), nullable=False), sa.Column("estimate_id", sa.Integer(), nullable=False), sa.Column("event_type", sa.String(length=40), nullable=False), sa.Column("actor_type", sa.String(length=40), nullable=False), sa.Column("actor_user_id", sa.Integer(), nullable=True), sa.Column("note", sa.Text(), nullable=True), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.ForeignKeyConstraint(["actor_user_id"], ["users.id"], ondelete="SET NULL"), sa.ForeignKeyConstraint(["estimate_id"], ["repair_estimates.id"], ondelete="CASCADE"), sa.PrimaryKeyConstraint("id"), ) op.create_index(op.f("ix_repair_estimate_events_actor_type"), "repair_estimate_events", ["actor_type"], unique=False) op.create_index(op.f("ix_repair_estimate_events_actor_user_id"), "repair_estimate_events", ["actor_user_id"], unique=False) op.create_index(op.f("ix_repair_estimate_events_estimate_id"), "repair_estimate_events", ["estimate_id"], unique=False) op.create_index(op.f("ix_repair_estimate_events_event_type"), "repair_estimate_events", ["event_type"], unique=False) for name, display_name, description, module in ESTIMATE_PERMISSIONS: op.execute( sa.text( """ INSERT INTO permissions (name, display_name, description, module) VALUES (:name, :display_name, :description, :module) ON CONFLICT (name) DO UPDATE SET display_name = excluded.display_name, description = excluded.description, module = excluded.module """ ).bindparams(name=name, display_name=display_name, description=description, module=module) ) role_permissions = { "administrator": [item[0] for item in ESTIMATE_PERMISSIONS], "management": ["repair_estimates.read", "repair_estimates.create", "repair_estimates.update", "repair_estimates.send"], "technician": ["repair_estimates.read", "repair_estimates.create", "repair_estimates.update", "repair_estimates.send"], "support": ["repair_estimates.read", "repair_estimates.send"], } for role_name, permission_names in role_permissions.items(): for permission_name in permission_names: op.execute( sa.text( """ INSERT INTO role_permissions (role_id, permission_id) SELECT roles.id, permissions.id FROM roles, permissions WHERE roles.name = :role_name AND permissions.name = :permission_name ON CONFLICT DO NOTHING """ ).bindparams(role_name=role_name, permission_name=permission_name) ) def downgrade() -> None: op.execute( sa.text( """ DELETE FROM role_permissions WHERE permission_id IN ( SELECT id FROM permissions WHERE module = 'repair_estimates' ) """ ) ) op.execute("DELETE FROM permissions WHERE module = 'repair_estimates'") op.drop_index(op.f("ix_repair_estimate_events_event_type"), table_name="repair_estimate_events") op.drop_index(op.f("ix_repair_estimate_events_estimate_id"), table_name="repair_estimate_events") op.drop_index(op.f("ix_repair_estimate_events_actor_user_id"), table_name="repair_estimate_events") op.drop_index(op.f("ix_repair_estimate_events_actor_type"), table_name="repair_estimate_events") op.drop_table("repair_estimate_events") op.drop_index(op.f("ix_repair_estimate_items_item_type"), table_name="repair_estimate_items") op.drop_index(op.f("ix_repair_estimate_items_estimate_id"), table_name="repair_estimate_items") op.drop_table("repair_estimate_items") op.drop_index(op.f("ix_repair_estimates_status"), table_name="repair_estimates") op.drop_index(op.f("ix_repair_estimates_repair_id"), table_name="repair_estimates") op.drop_index(op.f("ix_repair_estimates_estimate_number"), table_name="repair_estimates") op.drop_index(op.f("ix_repair_estimates_created_by_user_id"), table_name="repair_estimates") op.drop_table("repair_estimates")