spencerm/pesco-ncr
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
pesco-ncr/backend/alembic/versions/0001_initial_schema.py
ang3l12 dea316b113 Initial commit: PESCO NCR system 
Complete Non-Conformance Report system replacing the PowerApps/SharePoint
prototype: FastAPI + SQLAlchemy 2 (async) + Alembic + MySQL 8 backend,
React 18 + Vite + TypeScript + MUI frontend, Entra ID auth (MSAL / JWKS,
group-gated), Microsoft Graph delegated Mail.Send notifications (OBO),
six-stage workflow state machine with server-side enforcement, atomic
NCR-YYYY-NNNN numbering, attachments with camera capture, immutable
field-level audit trail, admin reopen, reports + CSV export, WeasyPrint
PDF traveler, Power BI reporting views + read-only DB user, documented
VISUAL ERP job-lookup stub, pytest suite (26 tests), docker-compose
deployment.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-13 11:41:22 -06:00

226 lines
9.0 KiB
Python
Raw Permalink Blame History

"""initial schema
Revision ID: 0001
Revises:
Create Date: 2026-07-13
"""
from alembic import op
import sqlalchemy as sa
revision = "0001"
down_revision = None
branch_labels = None
depends_on = None
MYSQL = {"mysql_charset": "utf8mb4", "mysql_collate": "utf8mb4_unicode_ci"}
def upgrade() -> None:
op.create_table(
"users",
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
sa.Column("entra_oid", sa.String(64), nullable=True, unique=True),
sa.Column("email", sa.String(255), nullable=False, unique=True, index=True),
sa.Column("display_name", sa.String(255), nullable=False),
sa.Column("employee_id", sa.String(64), nullable=True),
sa.Column("is_active", sa.Boolean(), nullable=False, server_default=sa.text("1")),
sa.Column("created_at", sa.DateTime(), nullable=False),
sa.Column("last_login_at", sa.DateTime(), nullable=True),
**MYSQL,
)
op.create_table(
"user_roles",
sa.Column(
"user_id",
sa.Integer(),
sa.ForeignKey("users.id", ondelete="CASCADE"),
primary_key=True,
),
sa.Column("role", sa.String(40), primary_key=True),
**MYSQL,
)
op.create_table(
"departments",
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
sa.Column("name", sa.String(100), nullable=False, unique=True),
sa.Column("is_active", sa.Boolean(), nullable=False, server_default=sa.text("1")),
**MYSQL,
)
op.create_table(
"deviation_categories",
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
sa.Column("name", sa.String(100), nullable=False, unique=True),
sa.Column("is_active", sa.Boolean(), nullable=False, server_default=sa.text("1")),
**MYSQL,
)
op.create_table(
"ncr_sequences",
sa.Column("year", sa.Integer(), primary_key=True, autoincrement=False),
sa.Column("last_seq", sa.Integer(), nullable=False, server_default=sa.text("0")),
**MYSQL,
)
op.create_table(
"ncrs",
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
sa.Column("ncr_number", sa.String(20), nullable=False),
sa.Column("ncr_year", sa.Integer(), nullable=False),
sa.Column("ncr_seq", sa.Integer(), nullable=False),
sa.Column("job_number", sa.String(100), nullable=False),
sa.Column("department_id", sa.Integer(), sa.ForeignKey("departments.id"), nullable=False),
sa.Column(
"deviation_category_id",
sa.Integer(),
sa.ForeignKey("deviation_categories.id"),
nullable=False,
),
sa.Column(
"disposition_authority_id", sa.Integer(), sa.ForeignKey("users.id"), nullable=False
),
sa.Column("deviation_detail", sa.Text(), nullable=False),
sa.Column("requester_id", sa.Integer(), sa.ForeignKey("users.id"), nullable=False),
sa.Column("stage", sa.String(30), nullable=False),
sa.Column("stage_entered_at", sa.DateTime(), nullable=False),
sa.Column("created_at", sa.DateTime(), nullable=False),
sa.Column("updated_at", sa.DateTime(), nullable=False),
sa.Column("qc_authority", sa.String(255), nullable=True),
sa.Column("work_order", sa.String(100), nullable=True),
sa.Column("disposition_notes", sa.Text(), nullable=True),
sa.Column("secondary_review_needed", sa.Boolean(), nullable=True),
sa.Column(
"operations_complete", sa.Boolean(), nullable=False, server_default=sa.text("0")
),
sa.Column("operations_completed_at", sa.DateTime(), nullable=True),
sa.Column(
"operations_completed_by_id", sa.Integer(), sa.ForeignKey("users.id"), nullable=True
),
sa.Column("qc_approval", sa.String(10), nullable=True),
sa.Column("inspection_notes", sa.Text(), nullable=True),
sa.Column("qc_closed", sa.Boolean(), nullable=False, server_default=sa.text("0")),
sa.Column("qc_closed_at", sa.DateTime(), nullable=True),
sa.Column("qc_closed_by_id", sa.Integer(), sa.ForeignKey("users.id"), nullable=True),
sa.Column("labor_cost", sa.Numeric(12, 2), nullable=True),
sa.Column("material_cost", sa.Numeric(12, 2), nullable=True),
sa.Column("service_cost", sa.Numeric(12, 2), nullable=True),
sa.Column("other_cost", sa.Numeric(12, 2), nullable=True),
sa.Column("costing_completed_at", sa.DateTime(), nullable=True),
sa.Column(
"costing_completed_by_id", sa.Integer(), sa.ForeignKey("users.id"), nullable=True
),
sa.Column("closed_at", sa.DateTime(), nullable=True),
sa.Column("closed_by_id", sa.Integer(), sa.ForeignKey("users.id"), nullable=True),
sa.UniqueConstraint("ncr_number", name="uq_ncrs_ncr_number"),
**MYSQL,
)
op.create_index("ix_ncrs_stage", "ncrs", ["stage"])
op.create_index("ix_ncrs_job_number", "ncrs", ["job_number"])
op.create_index("ix_ncrs_created_at", "ncrs", ["created_at"])
op.create_table(
"ncr_secondary_assignees",
sa.Column(
"ncr_id", sa.Integer(), sa.ForeignKey("ncrs.id", ondelete="CASCADE"), primary_key=True
),
sa.Column("user_id", sa.Integer(), sa.ForeignKey("users.id"), primary_key=True),
**MYSQL,
)
op.create_table(
"stage_transitions",
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
sa.Column(
"ncr_id", sa.Integer(), sa.ForeignKey("ncrs.id", ondelete="CASCADE"), nullable=False
),
sa.Column("from_stage", sa.String(30), nullable=True),
sa.Column("to_stage", sa.String(30), nullable=False),
sa.Column("action", sa.String(40), nullable=False),
sa.Column("acted_by_id", sa.Integer(), sa.ForeignKey("users.id"), nullable=False),
sa.Column("acted_at", sa.DateTime(), nullable=False),
sa.Column("note", sa.Text(), nullable=True),
**MYSQL,
)
op.create_index("ix_stage_transitions_ncr", "stage_transitions", ["ncr_id", "acted_at"])
op.create_table(
"job_info",
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
sa.Column(
"ncr_id",
sa.Integer(),
sa.ForeignKey("ncrs.id", ondelete="CASCADE"),
nullable=False,
unique=True,
),
sa.Column("part_id", sa.String(30), nullable=True),
sa.Column("part_description", sa.String(255), nullable=True),
sa.Column("customer_name", sa.String(100), nullable=True),
sa.Column("work_order_status", sa.String(20), nullable=True),
sa.Column("source", sa.String(20), nullable=False),
sa.Column("fetched_at", sa.DateTime(), nullable=False),
**MYSQL,
)
op.create_table(
"attachments",
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
sa.Column(
"ncr_id", sa.Integer(), sa.ForeignKey("ncrs.id", ondelete="CASCADE"), nullable=False
),
sa.Column("original_filename", sa.String(255), nullable=False),
sa.Column("stored_path", sa.String(300), nullable=False, unique=True),
sa.Column("content_type", sa.String(100), nullable=False),
sa.Column("size_bytes", sa.BigInteger(), nullable=False),
sa.Column("is_image", sa.Boolean(), nullable=False, server_default=sa.text("0")),
sa.Column("uploaded_by_id", sa.Integer(), sa.ForeignKey("users.id"), nullable=False),
sa.Column("uploaded_at", sa.DateTime(), nullable=False),
**MYSQL,
)
op.create_table(
"audit_log",
sa.Column("id", sa.BigInteger(), primary_key=True, autoincrement=True),
sa.Column(
"ncr_id", sa.Integer(), sa.ForeignKey("ncrs.id", ondelete="SET NULL"), nullable=True
),
sa.Column("user_id", sa.Integer(), sa.ForeignKey("users.id"), nullable=False),
sa.Column("created_at", sa.DateTime(), nullable=False),
sa.Column("action", sa.String(40), nullable=False),
sa.Column("field_name", sa.String(100), nullable=True),
sa.Column("old_value", sa.Text(), nullable=True),
sa.Column("new_value", sa.Text(), nullable=True),
sa.Column("detail", sa.String(500), nullable=True),
**MYSQL,
)
op.create_index("ix_audit_log_ncr", "audit_log", ["ncr_id", "created_at"])
op.create_index("ix_audit_log_created_at", "audit_log", ["created_at"])
op.create_table(
"app_settings",
sa.Column("key", sa.String(100), primary_key=True),
sa.Column("value", sa.String(500), nullable=False),
**MYSQL,
)
def downgrade() -> None:
for table in (
"app_settings",
"audit_log",
"attachments",
"job_info",
"stage_transitions",
"ncr_secondary_assignees",
"ncrs",
"ncr_sequences",
"deviation_categories",
"departments",
"user_roles",
"users",
):
op.drop_table(table)
Reference in New Issue View Git Blame Copy Permalink