"""ORM models for the Work Package Suite. Three tables: - sops one row per project SOP (the configuration baseline) - work_packages one row per IWP, optionally linked to a SOP - comments feedback / review comments from any page The full client document for a SOP or WP is kept verbatim in a JSON `data` column, with the most-queried fields promoted to real columns for listing and filtering. IDs are short strings (client- or server-generated) so the browser can upsert without round-tripping a sequence. NO relationship() DECLARATIONS, ON PURPOSE — and one consequence to know about. Every link here is a plain column plus a ForeignKey; nothing is navigable as `project.work_packages`. Queries are explicit selects, which suits an API that mostly reads one scoped list at a time and never wants a lazy load firing inside a response. The consequence: SQLAlchemy's unit of work derives FLUSH ORDER from relationships, not from ForeignKey metadata. With none declared it has no dependency edge to follow, so if you add a parent and its child in the SAME flush it may emit the child's INSERT first and the database will reject it. Both engines enforce foreign keys (Postgres always; SQLite since db.py sets `PRAGMA foreign_keys=ON`), so this is a real error, not a dev-only quirk. Call `db.flush()` after adding the parent — see `create_user` in app.py, which creates an account and its ProjectMember rows together. """ from datetime import datetime, timezone from typing import Optional from sqlalchemy import String, Boolean, Integer, DateTime, ForeignKey, Text, JSON, LargeBinary, UniqueConstraint from sqlalchemy.orm import Mapped, mapped_column from .db import Base def utcnow() -> datetime: return datetime.now(timezone.utc) class Project(Base): """A construction project — the top-level container. SOPs and Work Packages belong to a project so the suite can be used for many jobs at once.""" __tablename__ = "projects" id: Mapped[str] = mapped_column(String(40), primary_key=True) name: Mapped[str] = mapped_column(String(300), default="") number: Mapped[str] = mapped_column(String(100), default="", index=True) client: Mapped[str] = mapped_column(String(300), default="") division: Mapped[str] = mapped_column(String(200), default="") site: Mapped[str] = mapped_column(String(300), default="") sample: Mapped[bool] = mapped_column(Boolean, default=False) # Archived projects are hidden from every picker, switcher and search but kept # for the record — a finished job still has to be readable years later. Unlike # an archived work package they are also FROZEN read-only: the API refuses any # write to the project or to anything under it until an admin unarchives it. archived_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True, index=True) data: Mapped[dict] = mapped_column(JSON, default=dict) created_by: Mapped[str] = mapped_column(String(200), default="") created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, onupdate=utcnow) def summary(self) -> dict: return { "id": self.id, "name": self.name, "number": self.number, "client": self.client, "division": self.division, "site": self.site, "sample": self.sample, "archived_at": _iso(self.archived_at), "archived": self.archived_at is not None, "created_by": self.created_by, "created_at": _iso(self.created_at), "updated_at": _iso(self.updated_at), } def to_dict(self) -> dict: return {**self.summary(), "data": self.data or {}} class Sop(Base): __tablename__ = "sops" id: Mapped[str] = mapped_column(String(40), primary_key=True) project_id: Mapped[Optional[str]] = mapped_column( String(40), ForeignKey("projects.id", ondelete="CASCADE"), nullable=True, index=True ) name: Mapped[str] = mapped_column(String(300), default="") number: Mapped[str] = mapped_column(String(100), default="") complete: Mapped[bool] = mapped_column(Boolean, default=False) data: Mapped[dict] = mapped_column(JSON, default=dict) created_by: Mapped[str] = mapped_column(String(200), default="") created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, onupdate=utcnow) def summary(self) -> dict: return { "id": self.id, "project_id": self.project_id, "name": self.name, "number": self.number, "complete": self.complete, "created_by": self.created_by, "created_at": _iso(self.created_at), "updated_at": _iso(self.updated_at), } def to_dict(self) -> dict: return {**self.summary(), "data": self.data or {}} class WorkPackage(Base): __tablename__ = "work_packages" id: Mapped[str] = mapped_column(String(40), primary_key=True) project_id: Mapped[Optional[str]] = mapped_column( String(40), ForeignKey("projects.id", ondelete="CASCADE"), nullable=True, index=True ) sop_id: Mapped[Optional[str]] = mapped_column( String(40), ForeignKey("sops.id", ondelete="SET NULL"), nullable=True, index=True ) # parent_id links a discipline instance (WP01A) back to its master (WP01). parent_id: Mapped[Optional[str]] = mapped_column(String(40), nullable=True, index=True) number: Mapped[str] = mapped_column(String(120), default="") subject: Mapped[str] = mapped_column(String(400), default="") type: Mapped[str] = mapped_column(String(120), default="") status: Mapped[str] = mapped_column(String(40), default="Draft") # The accountable owner (a user id), for "My Work Packages" + assignment # notifications. Free-text `data.assignees`/`distribution` still hold the wider list. assignee_id: Mapped[Optional[str]] = mapped_column(String(40), nullable=True, index=True) issued_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) # Archived packages are hidden from the default lists/dashboard but kept for # the record (years-long projects accumulate hundreds of closed WPs). archived_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True, index=True) data: Mapped[dict] = mapped_column(JSON, default=dict) created_by: Mapped[str] = mapped_column(String(200), default="") created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, onupdate=utcnow) def summary(self) -> dict: return { "id": self.id, "project_id": self.project_id, "sop_id": self.sop_id, "parent_id": self.parent_id, "number": self.number, "subject": self.subject, "type": self.type, "status": self.status, "assignee_id": self.assignee_id, "issued_at": _iso(self.issued_at), "archived_at": _iso(self.archived_at), "archived": self.archived_at is not None, "created_by": self.created_by, "created_at": _iso(self.created_at), "updated_at": _iso(self.updated_at), } def to_dict(self) -> dict: return {**self.summary(), "data": self.data or {}} class User(Base): """A login account. NO PASSWORD IS STORED — D13 moved authentication to an LDAPS bind against the domain (see server/ldap_auth.py), and the `password_hash` column was dropped. `username` is the sAMAccountName people sign in with; an account is created on first successful sign-in if it does not already exist. `is_active` is LOCAL and overrides the directory: clearing it revokes access to this app without touching the domain account. Two independent notions of "role", deliberately separate: • role the PERMISSIONS role — what the account may do in the app. 'admin' | 'project_super_user' | 'project_admin' | 'project_user' (see auth.ROLES). • project_role the person's JOB FUNCTION on the project (Project Manager, Superintendent, QA/QC, …). Carries no permissions; it's what the SOP team pickers and notification routing read. """ __tablename__ = "users" id: Mapped[str] = mapped_column(String(40), primary_key=True) username: Mapped[str] = mapped_column(String(120), unique=True, index=True) email: Mapped[str] = mapped_column(String(200), default="") full_name: Mapped[str] = mapped_column(String(200), default="") role: Mapped[str] = mapped_column(String(20), default="project_user") # permissions role # Job function on the project — free text, offered from a suggested list. project_role: Mapped[str] = mapped_column(String(120), default="") # A PM or QA lead who belongs on every job shouldn't have to be ticked into each # new project by hand, so flagged accounts get a ProjectMember row the moment a # project is created. `auto_add_role` is the role they land with and shares # ProjectMember.role's value space: '' = inherit the account's own role, # otherwise 'project_admin' | 'project_user'. auto_add_projects: Mapped[bool] = mapped_column(Boolean, default=False) auto_add_role: Mapped[str] = mapped_column(String(20), default="") # Display preferences. Empty means "fall back to the app default, then to the # browser". A stored value follows the person between devices, which matters on # shared field tablets where the browser locale isn't theirs. locale: Mapped[str] = mapped_column(String(20), default="") # BCP47, e.g. en-US timezone: Mapped[str] = mapped_column(String(60), default="") # IANA, e.g. America/Chicago is_active: Mapped[bool] = mapped_column(Boolean, default=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, onupdate=utcnow) last_login_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) # Online-guessing throttle (see login()): consecutive failures + a lockout window. failed_attempts: Mapped[int] = mapped_column(Integer, default=0) locked_until: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) # Bumped to invalidate all existing sessions for this user. Password changes no # longer exist (D13), but a role change or a deactivation still has to take # effect on live sessions. The value is embedded in the JWT and re-checked on # every request. token_version: Mapped[int] = mapped_column(Integer, default=0) def to_dict(self) -> dict: """Public view of a user.""" return { "id": self.id, "username": self.username, "email": self.email, "full_name": self.full_name, "role": self.role, "project_role": self.project_role or "", "is_active": self.is_active, "auto_add_projects": bool(self.auto_add_projects), "auto_add_role": self.auto_add_role or "", "locale": self.locale or "", "timezone": self.timezone or "", "created_at": _iso(self.created_at), "last_login_at": _iso(self.last_login_at), } class ProjectMember(Base): """Which users may access which projects, and what they may do there. A user sees/operates on a project only if a row links them to it (admins bypass this entirely). One row per (user, project) pair. `role` is the permissions role ON THIS PROJECT: someone can be Project Admin on one job and a normal Project User on another, or a Project Super User (who administers that job's user accounts) on one job only. Empty means "inherit the account's own role" (User.role), which is how every existing row behaves. Values: '' | 'project_super_user' | 'project_admin' | 'project_user' (auth.PROJECT_SCOPED_ROLES) — never 'admin', which is app-wide by definition.""" __tablename__ = "project_members" __table_args__ = (UniqueConstraint("user_id", "project_id", name="uq_project_member"),) id: Mapped[str] = mapped_column(String(40), primary_key=True) user_id: Mapped[str] = mapped_column( String(40), ForeignKey("users.id", ondelete="CASCADE"), index=True ) project_id: Mapped[str] = mapped_column( String(40), ForeignKey("projects.id", ondelete="CASCADE"), index=True ) role: Mapped[str] = mapped_column(String(20), default="") # '' = inherit User.role created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) class LocationNode(Base): """One value in a project's Building / Floor / Sector taxonomy — CR-005. The taxonomy differs per project. On Micron, floors within B100 behave like separate buildings, so floor and sector are the unit of both execution and cost tracking; on another job "building" may be the only level that means anything. So it is configured once per SOP rather than hard-coded, and no real-world floor name appears anywhere in this repository. CODES, NOT DISPLAY STRINGS. `CR-018` rolls cost up by these, and a rollup keyed on a label breaks the day somebody fixes a typo in it. Two columns carry that: code this node's own slug among its siblings, derived once from the name it was imported with and then NEVER recomputed — renaming a node is a display change, which is exactly what makes rename safe for the work packages already pointing at it. path the full slug path from the root, '/'-joined and unique per project (`B100/L2/1P`). This is the grouping key and the value a work package stores. DEACTIVATE, NEVER DELETE. `active=False` hides a value from new work packages; every existing package referencing it still resolves its label, because the row is still there. Same rule as `CR-002`/`CR-016`: removal is expressed as a toggle, and the data is retained. """ __tablename__ = "location_nodes" __table_args__ = ( UniqueConstraint("project_id", "path", name="uq_location_path"), ) LEVELS = ("building", "floor", "sector") id: Mapped[str] = mapped_column(String(40), primary_key=True) project_id: Mapped[str] = mapped_column( String(40), ForeignKey("projects.id", ondelete="CASCADE"), index=True ) # Self-reference by id. No ForeignKey to its own table for the same reason the # rest of this file declares none — see the module docstring — and because a # self-referential FK plus SQLite's deferred-constraint behaviour makes a bulk # import fiddly for no gain. Orphans are prevented in the API, which is the # only writer. parent_id: Mapped[Optional[str]] = mapped_column(String(40), nullable=True, index=True) level: Mapped[str] = mapped_column(String(20), default="building") # building | floor | sector code: Mapped[str] = mapped_column(String(60), default="") # own slug path: Mapped[str] = mapped_column(String(200), default="", index=True) # full slug path name: Mapped[str] = mapped_column(String(200), default="") # display label active: Mapped[bool] = mapped_column(Boolean, default=True) sort: Mapped[int] = mapped_column(Integer, default=0) created_by: Mapped[str] = mapped_column(String(200), default="") created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, onupdate=utcnow) def to_dict(self) -> dict: return { "id": self.id, "project_id": self.project_id, "parent_id": self.parent_id, "level": self.level, "code": self.code, "path": self.path, "name": self.name, "active": bool(self.active), "sort": self.sort, "created_at": _iso(self.created_at), "updated_at": _iso(self.updated_at), } class Comment(Base): __tablename__ = "comments" id: Mapped[str] = mapped_column(String(40), primary_key=True) source: Mapped[str] = mapped_column(String(40), default="", index=True) # home_feedback | sop_step_comment | wp_review_comment sop_id: Mapped[Optional[str]] = mapped_column(String(40), nullable=True, index=True) wp_id: Mapped[Optional[str]] = mapped_column(String(40), nullable=True, index=True) step: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) author: Mapped[str] = mapped_column(String(200), default="") text: Mapped[str] = mapped_column(Text, default="") page: Mapped[str] = mapped_column(String(200), default="") extra: Mapped[dict] = mapped_column(JSON, default=dict) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) def to_dict(self) -> dict: return { "id": self.id, "source": self.source, "sop_id": self.sop_id, "wp_id": self.wp_id, "step": self.step, "author": self.author, "text": self.text, "page": self.page, "created_at": _iso(self.created_at), } class AuditLog(Base): """Append-only history: who changed what, when. Rows are written inside the same transaction as the change they describe (see server/app.py: log_event), so the trail can't drift from the data. `detail` holds a compact JSON summary of the change, e.g. {"from": "Scheduled", "to": "Issued"}. Not a ForeignKey to any entity on purpose — the log must survive the deletion of the thing it describes (you still want "who deleted WP01, and when").""" __tablename__ = "audit_log" id: Mapped[str] = mapped_column(String(40), primary_key=True) at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, index=True) actor: Mapped[str] = mapped_column(String(200), default="") # username who made the change action: Mapped[str] = mapped_column(String(60), default="", index=True) # created | updated | status_changed | issued | role_changed | ... entity_type: Mapped[str] = mapped_column(String(40), default="", index=True) # wp | sop | project | user entity_id: Mapped[str] = mapped_column(String(40), default="", index=True) project_id: Mapped[Optional[str]] = mapped_column(String(40), nullable=True, index=True) summary: Mapped[str] = mapped_column(String(400), default="") # human one-liner (e.g. the WP number/subject) detail: Mapped[dict] = mapped_column(JSON, default=dict) def to_dict(self) -> dict: return { "id": self.id, "at": _iso(self.at), "actor": self.actor, "action": self.action, "entity_type": self.entity_type, "entity_id": self.entity_id, "project_id": self.project_id, "summary": self.summary, "detail": self.detail or {}, } class AppSetting(Base): """Admin-editable application settings (feature flags, SMTP config, …) stored as key -> JSON value. Read/written via /api/settings (admin only). Secrets like the SMTP password are NOT stored here — they come from the environment.""" __tablename__ = "app_settings" key: Mapped[str] = mapped_column(String(80), primary_key=True) value: Mapped[dict] = mapped_column(JSON, default=dict) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, onupdate=utcnow) class MaterialItem(Base): """One line of a project's material list - D6 / T8.6, the CR-005 call made again: build the upload path now rather than wait for the master workbook. Deliberately small: description, unit, an optional code. NO inventory level, NO price, NO warehouse id - a project-scoped list the project uploaded is not the deferred parts catalog, and the moment a stock count appears here it has crossed the line IMPLEMENTATION.md section 7 draws. `active` rather than delete, same as everything else: a request already referencing a line must keep rendering it.""" __tablename__ = "material_items" id: Mapped[str] = mapped_column(String(40), primary_key=True) project_id: Mapped[str] = mapped_column(String(40), index=True) code: Mapped[str] = mapped_column(String(80), default="") description: Mapped[str] = mapped_column(String(300), default="") unit: Mapped[str] = mapped_column(String(20), default="") active: Mapped[bool] = mapped_column(Boolean, default=True) sort: Mapped[int] = mapped_column(Integer, default=0) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) def to_dict(self) -> dict: return {"id": self.id, "project_id": self.project_id, "code": self.code, "description": self.description, "unit": self.unit, "active": self.active, "sort": self.sort} class WpFile(Base): """CR-007 / D8: a drawing uploaded onto a work package. The BYTES live here, in the same database as everything else - settled Aug 18: a backup that excludes the drawings is a backup you cannot restore from. The limits are the D8 numbers: 5MB a file, PDFs and images, 2GB per project (80% warning). A meta copy (no bytes) is mirrored into the package's data["files"] by the server so the list is exportable and readable offline; that key is server-owned and survives client upserts.""" __tablename__ = "wp_files" id: Mapped[str] = mapped_column(String(40), primary_key=True) wp_id: Mapped[str] = mapped_column(String(40), index=True) project_id: Mapped[str] = mapped_column(String(40), index=True) name: Mapped[str] = mapped_column(String(300), default="") mime: Mapped[str] = mapped_column(String(100), default="") size: Mapped[int] = mapped_column(Integer, default=0) description: Mapped[str] = mapped_column(String(500), default="") data: Mapped[bytes] = mapped_column(LargeBinary, default=b"") uploaded_by: Mapped[str] = mapped_column(String(120), default="") created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) def to_dict(self) -> dict: # Meta only - the bytes go through GET /api/files/{id}, never through JSON. return { "id": self.id, "wp_id": self.wp_id, "project_id": self.project_id, "name": self.name, "mime": self.mime, "size": self.size, "description": self.description, "uploaded_by": self.uploaded_by, "created_at": self.created_at.isoformat() if self.created_at else None, } class Notification(Base): """Outbox for user notifications (an in-app record + an optional email). A row is written when something notable happens (e.g. a WP assignment); the email sender processes it only when email notifications are enabled AND SMTP is set — otherwise it's recorded as 'skipped'. See server/notify.py.""" __tablename__ = "notifications" id: Mapped[str] = mapped_column(String(40), primary_key=True) user_id: Mapped[str] = mapped_column(String(40), index=True) # recipient email: Mapped[str] = mapped_column(String(200), default="") kind: Mapped[str] = mapped_column(String(40), default="", index=True) # wp_assigned | … wp_id: Mapped[Optional[str]] = mapped_column(String(40), nullable=True) project_id: Mapped[Optional[str]] = mapped_column(String(40), nullable=True, index=True) subject: Mapped[str] = mapped_column(String(300), default="") body: Mapped[str] = mapped_column(Text, default="") link: Mapped[str] = mapped_column(String(500), default="") status: Mapped[str] = mapped_column(String(20), default="pending", index=True) # pending|sent|failed|skipped error: Mapped[str] = mapped_column(String(400), default="") created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, index=True) sent_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) def to_dict(self) -> dict: return { "id": self.id, "user_id": self.user_id, "email": self.email, "kind": self.kind, "wp_id": self.wp_id, "project_id": self.project_id, "subject": self.subject, "status": self.status, "error": self.error, "created_at": _iso(self.created_at), "sent_at": _iso(self.sent_at), } def _iso(dt: Optional[datetime]) -> Optional[str]: return dt.isoformat() if dt else None