n.siegfried/Project-SDE-WP-Suite
3
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity
Files
0652fa732dfe1f18227fca977297eef4da235bb8
Project-SDE-WP-Suite/server/models.py
Matt Mabrey 0652fa732d T11.1: UsageEvent model + migration (CR-019) 
New usage_events table, separate from audit_log - see the model docstring
for why. Verified: applies and downgrades cleanly on SQLite, and the
postgresql-dialect --sql render has no risky defaults (the BL-027 class
of defect). No app wiring yet - that's T11.2.
2026-09-23 11:23:14 -07:00

479 lines
26 KiB
Python
Raw Blame History

"""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 here or anywhere else — identity is
confirmed by Okta (OIDC), this app only decides what the account may do once
Okta has vouched for it (see server/okta_auth.py, server/auth.py, D15/D16).
`username` is what Okta's identity claim resolves to.
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 (e.g. on a password
# change). 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 — NEVER includes the password hash."""
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 UsageEvent(Base):
"""CR-019: append-only record of who used the suite, when, and which tool —
navigation/session activity, not business mutations. Deliberately a SEPARATE
table from AuditLog rather than a new `action` value there: AuditLog answers
"who changed what" and is read by people auditing a specific record's
history; mixing in a `page_open` row for every authenticated page load
would make that trail noisy for its existing purpose. This table answers a
different question — "who is active, and on what" — and CR-019's admin
console reads from here, not from AuditLog.
Not a ForeignKey to `users`, matching AuditLog's own reasoning: a user who
is later removed should still show up in historical activity rather than
silently vanishing from it, and `D18`'s deprovisioning sync only ever sets
`is_active=False` — it never deletes a row — so this is defensive symmetry
rather than a live concern today.
Retention is indefinite (decided 2026-09-17, `decisions-2026-09-17.md`) —
nothing purges rows written here; that is a deliberate product decision,
not an oversight to fix later."""
__tablename__ = "usage_events"
id: Mapped[str] = mapped_column(String(40), primary_key=True)
at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, index=True)
username: Mapped[str] = mapped_column(String(200), default="", index=True)
project_id: Mapped[Optional[str]] = mapped_column(String(40), nullable=True, index=True)
# creator | wizard | field_view | dashboard | admin | directory | ...
tool: Mapped[str] = mapped_column(String(40), default="", index=True)
# page_open | login
event: Mapped[str] = mapped_column(String(40), default="", index=True)
detail: Mapped[dict] = mapped_column(JSON, default=dict)
def to_dict(self) -> dict:
return {
"id": self.id, "at": _iso(self.at), "username": self.username,
"project_id": self.project_id, "tool": self.tool, "event": self.event,
"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
Reference in New Issue View Git Blame Copy Permalink