n.siegfried/Project-SDE-WP-Suite
3
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity
Files
3f4cd7ac923139815a0cacff58a983ddb8afe5b7
Project-SDE-WP-Suite/server/app.py
Cody Schaefer 3f4cd7ac92 Make a refused sign-in visible in the log 
The reason a sign-in was refused was logged at INFO, in app.py and in
ldap_auth. Nothing in this app configures the root logger, and uvicorn
configures only its own - so an INFO record from wpsuite.* reaches no handler
and is discarded. The message existed and could not be read, in exactly the
situation it was written for: someone cannot sign in and the operator needs to
know whether the credential was wrong, the account is outside the required
group, or the group does not resolve.

Raised to WARNING on the three refusal paths:

  app.py     "sign-in refused for 'x' (not_in_group: not in CN=...)"
  ldap_auth  "bind refused for 'x': 52e (bad password)"
  ldap_auth  "bind succeeded for 'x' but the account is NOT in 'CN=...'"

Left at INFO: provisioning an account, and normalising an address to a
sAMAccountName. Those are narrative, not diagnostic.

Config faults were already ERROR and were always visible, which is why the
503 path could be diagnosed and the 401 path could not.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-24 10:07:17 -05:00

3521 lines
167 KiB
Python
Raw Blame History

"""Work Package Suite API.
A small FastAPI service that stores project SOPs, Work Packages, and comments
in SQL (PostgreSQL in production; SQLite for local dev). NGINX serves the static
site and proxies /api/ here.
Run (dev): uvicorn server.app:app --reload --port 8000
Run (prod): gunicorn -k uvicorn.workers.UvicornWorker -b 127.0.0.1:8000 server.app:app
Interactive docs: http://<host>/api/docs
"""
import base64
import logging
from contextlib import asynccontextmanager
import os
import re
import uuid
from datetime import timedelta, timezone
from time import monotonic
from typing import Any, Optional
from urllib.parse import urlparse
from fastapi import FastAPI, Depends, HTTPException, Query, Request, Response, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import select, delete, func
from sqlalchemy.orm import Session
from .db import Base, engine, get_db
from . import models, auth, notify, ldap_auth, assets_db
# Same naming as the other modules' loggers (wpsuite.auth / .ldap / .notify), so a
# deployment can raise the level on one subsystem without raising it on all of them.
log = logging.getLogger("wpsuite.api")
# Schema management:
# • Local dev (SQLite) auto-creates tables for a zero-config run.
# • Production (Postgres) owns its schema through Alembic migrations, which run
# at container start (`alembic upgrade head`, see Dockerfile / DEPLOYMENT.md).
# We must NOT create_all there, or it would race/collide with the migration.
if engine.dialect.name == "sqlite":
Base.metadata.create_all(bind=engine)
# Interactive docs are handy in dev but hand an attacker the full API map in prod,
# so enable them only on the SQLite dev fallback (production runs on Postgres).
_docs_enabled = engine.dialect.name == "sqlite"
@asynccontextmanager
async def _lifespan(_app):
"""Say, once, whether anyone can sign in at all.
D13 removed the local password path and left no break-glass, so a broken LDAP
configuration and a forgotten password look identical at the login box. This
line is what tells an operator which one they have, and DEPLOYMENT.md,
DEPLOY-login-portal.md and server/README.md all send people here first:
docker compose logs api | grep -i "LDAP auth"
Configuration only — it opens no connection and binds nothing, so startup stays
fast and cannot be made to hang by an unreachable domain controller. Use
`ldap_auth.selftest()` for a reachability check; it validates the certificate
without binding, so it cannot contribute to a lockout either.
"""
log.info("%s", ldap_auth.describe())
yield
app = FastAPI(
lifespan=_lifespan,
title="Work Package Suite API",
docs_url="/api/docs" if _docs_enabled else None,
redoc_url=None,
openapi_url="/api/openapi.json" if _docs_enabled else None,
)
# Same-origin in production (NGINX), so CORS is normally unnecessary. For
# cross-origin local dev, set CORS_ORIGINS="http://localhost:5500,..."
# allow_credentials is required so the browser sends the session cookie.
_origins = [o for o in os.getenv("CORS_ORIGINS", "").split(",") if o]
if _origins:
app.add_middleware(
CORSMiddleware, allow_origins=_origins, allow_credentials=True,
allow_methods=["*"], allow_headers=["*"], expose_headers=["X-Total-Count"],
)
# ── Authentication gate ────────────────────────────────────────────────────────
# Every /api/ data route requires a valid session cookie. Login, health, and the
# docs are exempt (see auth._needs_auth). This is the real security boundary —
# the static pages are only client-side guarded for UX. OPTIONS (CORS preflight)
# is always allowed so the browser can negotiate before sending credentials.
_UNSAFE_METHODS = {"POST", "PUT", "PATCH", "DELETE"}
def _csrf_ok(request: Request) -> bool:
"""CSRF defense-in-depth behind SameSite=Lax: when the browser sends an Origin
on a state-changing request, it must be same-origin (or an allowed CORS origin).
Non-browser clients (no Origin header) are unaffected."""
origin = request.headers.get("origin")
if not origin:
return True
if _origins and origin in _origins:
return True
try:
return urlparse(origin).netloc == request.headers.get("host", "")
except Exception:
return False
@app.middleware("http")
async def auth_gate(request: Request, call_next):
path = request.url.path
method = request.method
if method != "OPTIONS" and auth._needs_auth(path):
if not auth.is_request_authenticated(request):
return JSONResponse(status_code=401, content={"detail": "Not authenticated"})
if method in _UNSAFE_METHODS and path.startswith("/api/") and not _csrf_ok(request):
return JSONResponse(status_code=403, content={"detail": "Cross-origin request rejected"})
return await call_next(request)
def gen_id(prefix: str) -> str:
return f"{prefix}_{uuid.uuid4().hex[:12]}"
# Clients may supply their own resource ids (offline-first). Constrain them to a
# safe charset so an id can never carry HTML/JS that a UI might place in markup.
_ID_RE = re.compile(r"^[A-Za-z0-9_-]{1,40}$")
def check_id(v: Optional[str]) -> None:
if v and not _ID_RE.match(v):
raise HTTPException(status_code=400, detail="Invalid id format")
# ── Per-project access control ─────────────────────────────────────────────────
# A non-admin user may only touch projects they're a member of (project_members).
# Admins bypass all of this. Resources with no project_id (legacy/orphan) are not
# gated. List endpoints are scoped to accessible projects; single-resource and
# mutating endpoints raise 403 on no access.
def accessible_project_ids(db: Session, user: "models.User"):
"""Return the set of project ids the user may access, or None for 'all' (admin)."""
if auth.is_admin(user):
return None
rows = db.scalars(
select(models.ProjectMember.project_id).where(models.ProjectMember.user_id == user.id)
).all()
return set(rows)
def require_project_access(db: Session, user: "models.User", project_id: Optional[str]) -> None:
if auth.is_admin(user):
return
if not project_id:
# Non-admins may not read/mutate resources with no project assignment
# (orphan/legacy rows); only admins can touch project-less data.
raise HTTPException(status_code=403, detail="This resource is not assigned to a project you can access")
ok = db.scalar(
select(models.ProjectMember.id).where(
(models.ProjectMember.user_id == user.id)
& (models.ProjectMember.project_id == project_id)
)
)
if not ok:
raise HTTPException(status_code=403, detail="You don't have access to this project")
def effective_role(db: Session, user: "models.User", project_id: Optional[str]) -> str:
"""The user's permissions role ON THIS PROJECT.
An app admin is admin everywhere. Otherwise a membership row may carry its own
role — so a PM on one job can be a plain Project User on another — and an empty
membership role falls back to the account's own role."""
if auth.is_admin(user):
return auth.ROLE_ADMIN
if project_id:
row = db.scalars(
select(models.ProjectMember).where(
(models.ProjectMember.user_id == user.id)
& (models.ProjectMember.project_id == project_id)
)
).first()
if row and (row.role or "").strip():
return auth.normalize_role(row.role)
return auth.normalize_role(user.role)
def require_project_admin(db: Session, user: "models.User", project_id: Optional[str],
what: str = "this action") -> None:
"""Destructive / baseline-changing operations: deleting a work package or a
project, and editing a SOP that has already been completed. Requires project
access AND Project Admin *on that project*."""
require_project_access(db, user, project_id)
if effective_role(db, user, project_id) not in (
auth.ROLE_ADMIN, auth.ROLE_PROJECT_SUPER, auth.ROLE_PROJECT_ADMIN,
):
raise HTTPException(
status_code=403,
detail=f"{what} requires the Project Admin role on this project",
)
def require_project_writable(db: Session, user_or_none, project_id: Optional[str],
what: str = "This change") -> None:
"""An archived project is frozen: everything on it stays readable, nothing on it
may be written. Every mutating path that lands on a project goes through here —
saving a SOP or a package, issuing, status changes, deletes, comments.
It is 409 and not 403 on purpose. Nobody lacks a permission here: the state of
the project is the objection, and even an admin gets refused until they unarchive
it — hence the caller identity is accepted for symmetry with the other guards but
deliberately unused. 409 also matters offline: the browser outbox in
html/project-data.js retires any 4xx op instead of retrying it forever, so an edit
queued before the archive dies quietly rather than looping against a frozen job."""
if not project_id:
return
proj = db.get(models.Project, project_id)
if proj is None:
# The project this write targets is gone — usually an outbox op queued before
# someone deleted the job. Refusing here is what keeps it a clean 409 instead
# of a foreign-key violation surfacing as a 500: the row could never be
# inserted anyway now that both engines enforce their FKs (see db.py). 409
# also matters because project-data.js retires a 4xx op and would retry a 5xx
# forever, so this is the difference between one quiet failure and a loop.
raise HTTPException(
status_code=409,
detail=f"{what} — this project no longer exists. It was deleted, so there is "
f"nothing to save it against.",
)
if proj.archived_at is not None:
raise HTTPException(
status_code=409,
detail=(f"{what} — this project is archived (read-only). An administrator "
f"can unarchive it from the admin console."),
)
def scope_to_access(stmt, column, db: Session, user: "models.User"):
"""Restrict a SELECT to the user's accessible projects (no-op for admins)."""
ids = accessible_project_ids(db, user)
if ids is None:
return stmt
return stmt.where(column.in_(ids))
def grant_project_access(db: Session, user_id: str, project_id: str, role: str = "") -> bool:
"""Add a (user, project) membership if it isn't already there, carrying an
optional per-project role override ('' = inherit the account's own). Returns True
only when a row was actually added, so a caller can report what it did. Does NOT
commit — the caller owns the transaction."""
exists = db.scalar(
select(models.ProjectMember.id).where(
(models.ProjectMember.user_id == user_id)
& (models.ProjectMember.project_id == project_id)
)
)
if exists:
return False
db.add(models.ProjectMember(id=gen_id("pm"), user_id=user_id, project_id=project_id, role=role))
return True
def add_default_members(db: Session, project_id: str, actor) -> list[str]:
"""Put the standing default members onto a brand-new project.
Some people belong on every job the moment it exists — the PM who runs them all,
the QC lead — and flagging their account (users.auto_add_projects) is how an admin
says that once instead of remembering it at every project creation. App admins are
skipped because they already reach every project, an inactive account is not
revived by a new job, and an existing membership is never duplicated or
overwritten. Returns the usernames added; does NOT commit, matching
grant_project_access."""
rows = db.scalars(
select(models.User).where(
(models.User.auto_add_projects.is_(True)) & (models.User.is_active.is_(True))
).order_by(models.User.username)
).all()
added = []
for u in rows:
if auth.is_admin(u):
continue
if grant_project_access(db, u.id, project_id, (u.auto_add_role or "").strip()):
added.append(u.username)
if added:
# One line for the batch, not one per person — this is a single automatic act.
log_event(db, actor, "project_access_granted", "project", project_id,
project_id=project_id, summary=f"{len(added)} default member(s) added",
detail={"users": added, "reason": "auto_add_projects"})
return added
# ── User-administration scope ──────────────────────────────────────────────────
# User administration used to be one thing: an app admin did all of it. It is now
# two, because a project admin has to be able to staff their own job without an app
# admin on the phone. An app admin still manages every account; a PROJECT SUPER USER
# manages the accounts on the projects they hold that role on.
#
# Three questions, deliberately separate, because they have different answers:
# managed_project_ids which projects do I administer the users of?
# visible_user_ids whose entry may I SEE in the directory?
# manage_user_problem may I change this account? (much narrower than seeing it)
def managed_project_ids(db: Session, caller: "models.User") -> Optional[set[str]]:
"""Projects where `caller` may administer user accounts. None means every project
(an app admin). Read per-membership so the per-project override decides: a super
user demoted to plain member on one job does not administer its users, and an
ordinary account made super user on one job does administer that one."""
if auth.is_admin(caller):
return None
rows = db.scalars(
select(models.ProjectMember).where(models.ProjectMember.user_id == caller.id)
).all()
account_role = auth.normalize_role(caller.role)
out = set()
for r in rows:
role = auth.normalize_role(r.role) if (r.role or "").strip() else account_role
if role == auth.ROLE_PROJECT_SUPER:
out.add(r.project_id)
return out
def is_user_manager(db: Session, user: "models.User") -> bool:
"""May this account administer users at all? THE one definition — derived from the
managed set, never from the account role alone, because the super-user role can be
held on a single project (ProjectMember.role) by an otherwise ordinary account.
Falls out of it that a super user with no project memberships manages nobody,
which is right: the authority comes from the jobs, not the job title."""
managed = managed_project_ids(db, user)
return managed is None or bool(managed)
def require_user_manager(user: "models.User" = Depends(auth.get_current_user),
db: Session = Depends(get_db)) -> "models.User":
"""First gate on the user-administration routes: does this caller administer the
users of ANY project? Which accounts they may then touch is a second, narrower
check per target — require_manage_user."""
if not is_user_manager(db, user):
raise HTTPException(
status_code=403,
detail="Managing user accounts requires the Administrator role, or Project "
"Super User on a project",
)
return user
def member_project_ids(db: Session, user_id: str) -> set[str]:
"""Every project this user has a membership row for (no admin shortcut — this is
the raw set, which is exactly what the scope checks need to reason about)."""
return set(db.scalars(
select(models.ProjectMember.project_id).where(models.ProjectMember.user_id == user_id)
).all())
def visible_user_ids(db: Session, caller: "models.User") -> Optional[set[str]]:
"""Whose directory entry `caller` may read. None means everyone (an app admin).
Anyone signed in may look up the people they actually work with — their own
projects' members — plus the app admins, who are on every project implicitly and
are who you go to when something needs unblocking. Nobody else: the directory
must not become a company-wide address book for a single-project subcontractor."""
if auth.is_admin(caller):
return None
ids = {caller.id}
mine = accessible_project_ids(db, caller) or set()
if mine:
ids |= set(db.scalars(
select(models.ProjectMember.user_id).where(models.ProjectMember.project_id.in_(mine))
).all())
ids |= set(db.scalars(
select(models.User.id).where(models.User.role == auth.ROLE_ADMIN)
).all())
return ids
def manage_user_problem(db: Session, caller: "models.User", target: "models.User",
cache: Optional[dict] = None) -> Optional[str]:
"""None if `caller` may make ACCOUNT-level changes to `target` (password, name,
permissions role, enable/disable, delete); otherwise the reason they may not, in
words the console can show verbatim.
An app admin may always. A super user may only when the account sits ENTIRELY
inside the projects they administer, and is not itself an admin or super user.
Both limits matter:
• Exclusive scope, because these changes are global. Resetting a password or
disabling an account reaches every project that person is on, so a super user
must not be able to reach into a job they don't run by way of a shared member.
• No admin/super targets, because otherwise the role could be used to take over
a peer's account and inherit their scope.
Project-scoped changes (adding someone to MY project, their role THERE) are not
account-level and are checked against `managed_project_ids` instead.
`cache` lets a caller judging a whole page of users hand in the two lookups this
needs ('managed', and 'members' as {user_id: {project_id}}) so the verdict for
thirty rows costs two queries instead of sixty. The rule itself lives only here."""
if auth.is_admin(caller):
return None
cache = cache if cache is not None else {}
managed = cache.get("managed")
if managed is None:
managed = cache["managed"] = managed_project_ids(db, caller) or set()
if not managed:
return ("You don't administer the users of any project — that needs the Project "
"Super User role on the project")
if auth.normalize_role(target.role) in (auth.ROLE_ADMIN, auth.ROLE_PROJECT_SUPER):
return "Only an application administrator can change an Administrator or Project Super User account"
members = cache.get("members")
theirs = members.get(target.id, set()) if members is not None else member_project_ids(db, target.id)
if not theirs:
return ("This account isn't on any project, so only an application administrator "
"can change it")
outside = theirs - managed
if outside:
return (f"{target.username} is also on {len(outside)} project(s) you don't administer — "
"account changes there have to come from an application administrator. "
"You can still change their access and role on your own projects.")
return None
def require_manage_user(db: Session, caller: "models.User", target: "models.User") -> None:
problem = manage_user_problem(db, caller, target)
if problem:
raise HTTPException(status_code=403, detail=problem)
def require_see_user(db: Session, caller: "models.User", target: "models.User") -> None:
"""404, not 403: whether an account exists outside your projects is itself not
yours to learn, and a 403 would confirm the username."""
visible = visible_user_ids(db, caller)
if visible is not None and target.id not in visible:
raise HTTPException(status_code=404, detail="User not found")
def grantable_roles(caller: "models.User") -> tuple:
"""Permissions roles `caller` may hand out. A super user may staff their job with
project admins and project users — never another admin or super user, which is the
line that keeps the role from being a route to app-wide control."""
if auth.is_admin(caller):
return auth.ROLES
return (auth.ROLE_PROJECT_ADMIN, auth.ROLE_PROJECT_USER)
def load_target_user(db: Session, user_id: str) -> "models.User":
u = db.get(models.User, user_id)
if not u:
raise HTTPException(status_code=404, detail="User not found")
return u
# ── Audit trail ────────────────────────────────────────────────────────────────
def log_event(db: Session, actor, action: str, entity_type: str, entity_id: str,
project_id: Optional[str] = None, summary: str = "", detail: Optional[dict] = None) -> None:
"""Append an audit-trail row in the CURRENT transaction so it commits
atomically with the change it describes. `actor` may be a User or a username."""
who = actor.username if isinstance(actor, models.User) else (actor or "")
db.add(models.AuditLog(
id=gen_id("ev"), actor=who, action=action, entity_type=entity_type,
entity_id=entity_id or "", project_id=project_id, summary=(summary or "")[:400], detail=detail or {},
))
# ── Assignment ─────────────────────────────────────────────────────────────────
def require_assignable(db: Session, user_id: str, project_id: Optional[str]) -> None:
"""A WP can only be assigned to an active user who can access its project."""
u = db.get(models.User, user_id)
if not u or not u.is_active:
raise HTTPException(status_code=400, detail="Assignee is not a valid user")
if auth.is_admin(u):
return
ok = db.scalar(
select(models.ProjectMember.id).where(
(models.ProjectMember.user_id == user_id) & (models.ProjectMember.project_id == project_id)
)
)
if not ok:
raise HTTPException(status_code=400, detail="Assignee is not a member of this project")
def wp_link(db: Session, wp: "models.WorkPackage") -> str:
"""A link that opens THIS work package. X1: never the app root - a recipient
who has to hunt for the package after signing in stops opening the emails.
The creator has been its own document since T7.1 and boots ?wp= deep links;
a signed-out recipient rides login.html?next= straight back to it."""
base = (notify.get_settings(db).get("app_base_url") or "").rstrip("/")
path = f"/wp-creation-index.html?project={wp.project_id or ''}&wp={wp.id}"
return (base + path) if base else path
def wp_titled(wp: "models.WorkPackage") -> str:
"""Number — title, for a message body. Decided 2026-08-20: the title is
customer CONTEXT and may ride in mail; document CONTENT may not."""
t = (wp.subject or "").strip()
n = wp.number or "a work package"
return f"{n} — {t}" if t else n
def wp_where(wp: "models.WorkPackage") -> str:
"""Where the work happens, for a message body: the CR-004 structured
fields (stored as paths — stable, and readable to the people these mails
address), else the pre-CR-004 free text. Empty string when unset, and
callers drop the line entirely rather than mail 'Where: '."""
data = wp.data or {}
parts = [str(data.get(d) or "").strip() for d in LOCATION_DIMENSIONS]
parts = [p for p in parts if p]
return " / ".join(parts) if parts else str(data.get("location") or "").strip()
def _where_line(wp: "models.WorkPackage") -> str:
w = wp_where(wp)
return f"Where: {w}\n" if w else ""
def assign_body(assignee: "models.User", wp: "models.WorkPackage", actor: "models.User", link: str) -> str:
# Number, title and location — customer context, allowed since the
# 2026-08-20 decision (decisions-2026-08-20.md). Contents stay behind
# the link: no scope text, no descriptions, no attachments.
who = actor.full_name or actor.username
name = assignee.full_name or assignee.username
return (
f"Hi {name},\n\n"
f"{who} assigned you a work package: {wp_titled(wp)}.\n"
+ _where_line(wp) +
f"\nOpen the Work Package Suite to view and action it:\n{link}\n\n"
f"— This is an automated message from the Work Package Suite."
)
# ── Request bodies ───────────────────────────────────────────────────────────
class ProjectIn(BaseModel):
id: Optional[str] = None
name: str = ""
number: str = ""
client: str = ""
division: str = ""
site: str = ""
sample: bool = False
created_by: str = ""
data: dict[str, Any] = Field(default_factory=dict)
class SopIn(BaseModel):
id: Optional[str] = None
project_id: Optional[str] = None
name: str = ""
number: str = ""
complete: bool = False
created_by: str = ""
data: dict[str, Any] = Field(default_factory=dict)
class WpIn(BaseModel):
id: Optional[str] = None
project_id: Optional[str] = None
sop_id: Optional[str] = None
parent_id: Optional[str] = None
number: str = ""
subject: str = ""
type: str = ""
status: str = "Draft"
assignee_id: Optional[str] = None
created_by: str = ""
data: dict[str, Any] = Field(default_factory=dict)
class SettingsIn(BaseModel):
model_config = ConfigDict(extra="ignore")
email_enabled: Optional[bool] = None
smtp_host: Optional[str] = None
smtp_port: Optional[int] = None
smtp_use_tls: Optional[bool] = None
smtp_username: Optional[str] = None
from_addr: Optional[str] = None
from_name: Optional[str] = None
app_base_url: Optional[str] = None
bim_enabled: Optional[bool] = None
default_locale: Optional[str] = None
default_timezone: Optional[str] = None
class TestEmailIn(BaseModel):
to: Optional[str] = None
# CR-007 / D8: the drawing-upload numbers, as settled Aug 18. The ceiling is
# env-overridable so a test can drive the 80% warning and the refusal without
# writing two gigabytes; the DEFAULT is the decision.
FILE_MAX_BYTES = 5 * 1024 * 1024
FILE_PROJECT_CEILING = int(os.getenv("WP_FILE_PROJECT_CEILING", str(2 * 1024 * 1024 * 1024)))
FILE_ALLOWED_MIME_RE = re.compile(r"^(application/pdf|image/[a-z0-9.+-]+)$")
class FileUploadIn(BaseModel):
name: str = ""
mime: str = ""
description: str = ""
data_base64: str = ""
class FileDescIn(BaseModel):
description: str = ""
class StatusIn(BaseModel):
status: str
# CR-014: a rejection (Ready for QA -> In Progress) must say why. The upsert
# route carries the comment inside data.qaRejections; this route has no data,
# so it carries the comment here and the server appends the record itself.
comment: Optional[str] = None
class ArchiveIn(BaseModel):
archived: bool = True
class CommentIn(BaseModel):
# Tolerate any extra keys the feedback payload includes (timestamp, app, …).
model_config = ConfigDict(extra="allow")
source: Optional[str] = None
type: Optional[str] = None # client sends 'type'; treated as source
sop_id: Optional[str] = None
wp_id: Optional[str] = None
step: Optional[int] = None
author: Optional[str] = None
name: Optional[str] = None # home/SOP forms send 'name'
text: Optional[str] = None
page: Optional[str] = ""
# ── Health ───────────────────────────────────────────────────────────────────
@app.get("/api/health")
def health():
return {"ok": True}
# ── Authentication ─────────────────────────────────────────────────────────────
class LoginIn(BaseModel):
username: str
password: str
class NewUserIn(BaseModel):
# No password: D13 authenticates against the domain, so an administrator
# pre-creating an account only supplies identity and authorization. The person
# signs in with their Windows password, or is provisioned on first sign-in.
username: str
full_name: str = ""
email: str = ""
role: str = auth.ROLE_PROJECT_USER # permissions role — see auth.ROLES
project_role: str = "" # job function on the project (no permissions)
# Projects to put the new account on straight away. Optional for an app admin
# (who can assign later); REQUIRED for a super user, whose authority over an
# account comes from the projects it is on — see create_user.
project_ids: list[str] = Field(default_factory=list)
class ProjectRoleIn(BaseModel):
project_role: str = ""
class PreferencesIn(BaseModel):
# Empty string clears the preference (fall back to the app default, then the
# browser). None means "leave this one alone".
locale: Optional[str] = None
timezone: Optional[str] = None
class ActiveIn(BaseModel):
is_active: bool
class RoleIn(BaseModel):
role: str # permissions role — see auth.ROLES
class ProjectAssignIn(BaseModel):
project_ids: list[str] = Field(default_factory=list)
# Optional per-project permissions role, {project_id: role}. Omit or use '' to
# inherit the account's own role on that project.
roles: dict[str, str] = Field(default_factory=dict)
class AutoAddIn(BaseModel):
auto_add: bool
# Role to give this user on the projects they're auto-added to; '' inherits the
# account's own role, same value space as ProjectMember.role.
role: str = ""
# D13 / T10.2 — THIS NUMBER IS NOT A UX PREFERENCE, IT IS A SAFETY LIMIT.
#
# Failures are now LDAP binds, so every one counts against the DOMAIN account
# lockout policy. This estate's AD threshold is 5. The throttle below is
# per-process and the API runs 2 gunicorn workers, so a local limit of N lets up
# to 2N binds reach a domain controller: 2 x 2 = 4, one under the threshold.
#
# It defaulted to 5 before this task, which would have allowed up to 10 binds and
# locked the account out of WINDOWS — twice over — before the local lockout ever
# engaged. If you raise this, or add a worker, redo the arithmetic first.
LOGIN_MAX_ATTEMPTS = int(os.getenv("AUTH_MAX_ATTEMPTS", "2"))
LOGIN_LOCKOUT_MINUTES = int(os.getenv("AUTH_LOCKOUT_MINUTES", "15"))
# Pre-account throttle, keyed by USERNAME (not by client IP — an attacker rotating
# IPs would sail past an IP-keyed limit, and it is the domain account we are
# protecting, not this endpoint's capacity).
#
# The DB counter on `users` is shared across workers and persists, but it can only
# work for an account that already has a local row. Under D13 accounts are created
# on first successful login, so a REAL domain account can be hammered before any
# row exists — this dict covers exactly that window. Per-worker and lost on
# restart, which is why LOGIN_MAX_ATTEMPTS is halved rather than trusted.
_bind_attempts: dict[str, list[float]] = {}
_BIND_WINDOW_SECONDS = LOGIN_LOCKOUT_MINUTES * 60
def _bind_throttled(sam: str) -> bool:
"""True if this username has already used its bind budget in the window.
Checked BEFORE any call to the directory."""
now = monotonic()
key = (sam or "").strip().lower()
hits = [t for t in _bind_attempts.get(key, []) if now - t < _BIND_WINDOW_SECONDS]
_bind_attempts[key] = hits
if len(_bind_attempts) > 5000: # bound the dict on a long-lived worker
for k in [k for k, v in _bind_attempts.items()
if not v or now - max(v) > _BIND_WINDOW_SECONDS]:
_bind_attempts.pop(k, None)
return len(hits) >= LOGIN_MAX_ATTEMPTS
def _record_bind_failure(sam: str) -> None:
_bind_attempts.setdefault((sam or "").strip().lower(), []).append(monotonic())
def _clear_bind_failures(sam: str) -> None:
_bind_attempts.pop((sam or "").strip().lower(), None)
def _match_directory_account(db: Session, result) -> Optional[models.User]:
"""Find the local row for a directory identity — D13 criterion 4.
Matched on `sAMAccountName` OR the directory's `mail`, because existing accounts
were created by hand with `manage_users.py` and some were typed as short logon
names while others were typed as email addresses. Matching on both is what keeps
an existing admin's role instead of handing them a second, default-role account.
`auth.find_user` already compares case-insensitively against username AND email,
so each call covers two columns; the second call is for the case where the local
username is the person's address and the directory only told us their sAMAccountName.
"""
user = auth.find_user(db, result.sam)
if user is None and result.mail:
user = auth.find_user(db, result.mail)
if user is not None:
log.info("matched directory identity %r to existing local account %r by mail",
result.sam, user.username)
return user
def _provision_from_directory(db: Session, result) -> models.User:
"""Create a local account for someone who just authenticated and has no row.
Lands at `project_user` with NO project memberships. That is least privilege and
it is deliberate, but it means the person signs in successfully into an empty
app until an admin grants access — so it is written to the audit log rather than
happening silently. `auto_add_projects` cannot help here: it is evaluated when a
PROJECT is created, to mark who joins every new job, and cannot retroactively add
a new account to jobs that already exist.
"""
u = models.User(
id=gen_id("user"),
username=result.sam,
email=result.mail or "",
full_name=result.full_name or "",
role=auth.ROLE_PROJECT_USER,
)
db.add(u)
db.flush() # see the flush-order note in models.py's docstring
log_event(db, u.username, "user_provisioned", "user", u.id, summary=u.username,
detail={"source": "directory", "role": u.role, "upn": result.upn,
"projects": 0, "note": "created on first successful sign-in"})
log.info("provisioned local account %r from the directory at role %r with no "
"project access — an admin must grant access before they see anything",
u.username, u.role)
return u
@app.post("/api/auth/login")
def login(body: LoginIn, request: Request, response: Response, db: Session = Depends(get_db)):
"""Authenticate against the domain (D13 / T10.2) and set the session cookie.
The suite stores no passwords: this is an LDAPS simple bind as
`<sAMAccountName>@prime.local`, and a successful bind IS the authentication.
See server/ldap_auth.py for the transport and its two hard guards.
ORDER MATTERS HERE. Every throttle check runs BEFORE the directory is touched,
because a failed bind counts against the DOMAIN lockout policy — so this
endpoint must not be usable to lock a colleague out of Windows. Nothing below
reaches `ldap_auth.verify` until the local budget has been checked twice: once
against the persistent per-account counter, once against the pre-account
window that covers usernames with no local row yet.
Three outcomes, deliberately distinguished:
401 the credential was rejected, or the account is not in the required
group. One generic message for every case — the response must never
reveal whether an account exists (see `_ERR49` in ldap_auth: the useful
detail goes to the log).
403 the local account exists and is disabled. Independent of the directory.
503 OUR fault — LDAP unconfigured, unreachable, untrusted, or the required
group does not resolve. D13 left no password fallback, so this must not
masquerade as 401: "your password is wrong" sends people hunting for a
password they no longer have, while the real problem is a broken deploy.
"""
now = models.utcnow()
sam = ldap_auth.normalize_username(body.username)
if not sam or not (body.password or "").strip():
# No directory call for empty input. ldap_auth.verify guards this too; the
# duplication is intentional, since an empty password would otherwise be an
# anonymous bind and anonymous binds SUCCEED.
raise HTTPException(status_code=401, detail="Invalid username or password")
user = auth.find_user(db, sam)
# ── throttle 1: the persistent per-account lockout ────────────────────────
locked = user.locked_until if user else None
if locked is not None and locked.tzinfo is None:
locked = locked.replace(tzinfo=timezone.utc) # SQLite returns naive datetimes
if locked is not None and locked > now:
raise HTTPException(status_code=429, detail="Too many failed attempts. Try again later.")
# ── throttle 2: the pre-account window ────────────────────────────────────
if _bind_throttled(sam):
log.warning("refusing to bind for %r — local attempt budget (%d) spent; "
"protecting the domain account from lockout", sam, LOGIN_MAX_ATTEMPTS)
raise HTTPException(status_code=429, detail="Too many failed attempts. Try again later.")
# ── the directory ─────────────────────────────────────────────────────────
settings = notify.get_settings(db)
result = ldap_auth.verify(sam, body.password,
required_group=settings.get("ldap_required_group"))
if result.is_config_problem:
# Ours, not theirs. Never counted against the user, never a 401.
log.error("sign-in unavailable — %s: %s", result.reason, result.detail)
raise HTTPException(status_code=503,
detail="Sign-in is temporarily unavailable. Contact IT.")
if not result.ok:
# WARNING, not INFO: this is the line an operator needs when someone
# cannot sign in, and nothing configures the root logger — under plain
# uvicorn an INFO record from wpsuite.* goes nowhere, so the reason was
# invisible in exactly the situation it exists for.
log.warning("sign-in refused for %r (%s: %s)", sam, result.reason, result.detail)
_record_bind_failure(sam)
if user:
user.failed_attempts = (user.failed_attempts or 0) + 1
if user.failed_attempts >= LOGIN_MAX_ATTEMPTS:
user.locked_until = now + timedelta(minutes=LOGIN_LOCKOUT_MINUTES)
user.failed_attempts = 0
log_event(db, user.username, "login_locked", "user", user.id,
summary=user.username,
detail={"minutes": LOGIN_LOCKOUT_MINUTES, "reason": result.reason})
db.commit()
raise HTTPException(status_code=401, detail="Invalid username or password")
# ── authenticated ─────────────────────────────────────────────────────────
_clear_bind_failures(sam)
# The pre-bind lookup used the typed name only. Now that the directory has told
# us the account's real sAMAccountName and mail, re-match on both (T10.4).
if user is None:
user = _match_directory_account(db, result)
provisioned = user is None
if provisioned:
user = _provision_from_directory(db, result)
else:
# NEVER touch `role` here. An existing admin stays an admin — that is D13
# criterion 4, and it is the whole reason this branch is separate from the
# one above. Fill in identity fields only where they are empty locally, so a
# name deliberately set in the console is not overwritten by the directory.
if not user.full_name and result.full_name:
user.full_name = result.full_name
if not user.email and result.mail:
user.email = result.mail
if not user.is_active:
# Checked after provisioning so a brand-new account (is_active defaults True)
# is not caught by it, and after the role branch so a disabled admin is still
# refused. Local state overrides the directory: disabling here is how you
# revoke access to THIS app without touching the domain account.
raise HTTPException(status_code=403, detail="Account is disabled")
user.failed_attempts = 0
user.locked_until = None
user.last_login_at = now
db.commit()
db.refresh(user)
token = auth.create_token(user)
auth.set_session_cookie(response, request, token)
return {"user": user.to_dict()}
@app.post("/api/auth/logout")
def logout(response: Response):
auth.clear_session_cookie(response)
return {"ok": True}
@app.get("/api/auth/me")
def whoami(user: models.User = Depends(auth.get_current_user)):
"""Who is logged in. The frontend guard calls this on every page load.
`role` is normalized here so no page has to know that a pre-roles account stores
'user' where it now means 'project_user'."""
return {"user": {**user.to_dict(), "role": auth.normalize_role(user.role)}}
# ── Display preferences (self-service) ─────────────────────────────────────────
_LOCALE_RE = re.compile(r"^[A-Za-z]{2,3}(-[A-Za-z0-9]{2,8}){0,3}$")
def valid_timezone(tz: str) -> bool:
"""True if this is a real IANA zone name on this machine."""
try:
from zoneinfo import ZoneInfo
ZoneInfo(tz)
return True
except Exception: # noqa: BLE001 — unknown key, missing tzdata, bad type
return False
@app.get("/api/timezones")
def list_timezones(_user: models.User = Depends(auth.get_current_user)):
"""IANA zone names for the preferences picker, so the list matches what the
server will actually accept."""
try:
from zoneinfo import available_timezones
return sorted(available_timezones())
except Exception: # noqa: BLE001 — no tzdata: let the client fall back
return []
@app.post("/api/auth/preferences")
def set_preferences(body: PreferencesIn, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
"""A user's own locale / timezone. Empty string clears the preference; the app
default (admin console) applies next, then the browser's own settings."""
if body.locale is not None:
loc = body.locale.strip()
if loc and not _LOCALE_RE.match(loc):
raise HTTPException(status_code=400, detail="Locale must be a language tag like 'en-US' or 'es'.")
user.locale = loc[:20]
if body.timezone is not None:
tz = body.timezone.strip()
if tz and not valid_timezone(tz):
raise HTTPException(status_code=400, detail="Unknown time zone. Pick one from the list.")
user.timezone = tz[:60]
db.commit()
db.refresh(user)
return {"user": user.to_dict()}
# ── User administration ─────────────────────────────────────────────────────────
# Two kinds of caller reach these routes: an app admin, who manages every account,
# and a Project Super User, who manages the accounts on the projects they administer.
# Every route therefore asks TWO questions — does this role carry user administration
# (require_user_manager), and may it touch THIS account (require_manage_user) —
# and the read route asks a third, wider one (visible_user_ids) because looking a
# colleague up is not the same as being able to change them.
def directory_entry(db: Session, u: "models.User", caller: "models.User",
counts: Optional[dict] = None, cache: Optional[dict] = None) -> dict:
"""One row of the user directory, cut to what `caller` is entitled to see.
A manager gets the administrative record (last login, the auto-add flags, and a
`manageable` verdict with the reason when it's no). Everyone else gets the contact
card only — a project user has no business reading their colleagues' login history
out of a page whose job is "who is on this project and how do I reach them"."""
if not is_user_manager(db, caller):
return {
"id": u.id, "username": u.username, "full_name": u.full_name, "email": u.email,
"role": auth.normalize_role(u.role), "project_role": u.project_role or "",
"is_active": u.is_active, "manageable": False,
}
problem = manage_user_problem(db, caller, u, cache)
n = None if counts is None else counts.get(u.id, 0)
return {
**u.to_dict(),
"role": auth.normalize_role(u.role),
"manageable": problem is None,
"manage_blocked_reason": problem or "",
"project_count": n,
}
@app.get("/api/auth/users")
def list_users(user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
"""The user directory, scoped to the caller. An app admin sees every account; a
project member sees the people on their own projects (plus the app admins)."""
visible = visible_user_ids(db, user)
stmt = select(models.User).order_by(models.User.username)
if visible is not None:
stmt = stmt.where(models.User.id.in_(visible))
rows = db.scalars(stmt).all()
# One pass over the membership table serves both the project-access count and the
# per-row "may I manage this account" verdict. The old console fetched the counts
# with one HTTP request per user.
counts, cache = None, None
if is_user_manager(db, user):
members: dict[str, set] = {}
for uid, pid in db.execute(
select(models.ProjectMember.user_id, models.ProjectMember.project_id)
).all():
members.setdefault(uid, set()).add(pid)
counts = {uid: len(pids) for uid, pids in members.items()}
cache = {"members": members}
return [directory_entry(db, u, user, counts, cache) for u in rows]
@app.get("/api/auth/user-scope")
def user_scope(user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
"""What the signed-in account may do on the user directory page, so the page can
render the right controls instead of guessing at the rules and drawing buttons
that 403. Advisory only — every route re-checks server-side."""
managed = managed_project_ids(db, user)
if managed is None:
rows = db.scalars(select(models.Project).order_by(models.Project.name)).all()
else:
rows = db.scalars(
select(models.Project).where(models.Project.id.in_(managed)).order_by(models.Project.name)
).all() if managed else []
manager = managed is None or bool(managed)
return {
"can_manage_users": manager,
"scope": "all" if managed is None else "projects",
"role": auth.normalize_role(user.role),
"grantable_roles": list(grantable_roles(user)) if manager else [],
"grantable_project_roles": list(
auth.PROJECT_SCOPED_ROLES if auth.is_admin(user)
else (auth.ROLE_PROJECT_ADMIN, auth.ROLE_PROJECT_USER)
) if manager else [],
"role_labels": auth.ROLE_LABELS,
"project_roles": list(auth.PROJECT_ROLES),
"managed_projects": [{"id": p.id, "name": p.name, "number": p.number,
"archived": p.archived_at is not None} for p in rows],
}
@app.post("/api/auth/users")
def create_user(body: NewUserIn, actor: models.User = Depends(require_user_manager), db: Session = Depends(get_db)):
allowed = grantable_roles(actor)
if body.role not in allowed:
raise HTTPException(status_code=400, detail=f"role must be one of {', '.join(allowed)}")
if auth.find_user(db, body.username):
raise HTTPException(status_code=409, detail="A user with that username already exists")
managed = managed_project_ids(db, actor)
requested = [p for p in dict.fromkeys(body.project_ids) if p]
for pid in requested:
check_id(pid)
if managed is not None:
# A super user's authority over an account is derived from the projects that
# account is on. Creating one with no project — or on a job they don't run —
# would either produce an account they instantly cannot manage, or reach into
# someone else's job. Both are refused rather than silently narrowed.
if not requested:
raise HTTPException(
status_code=400,
detail="Choose at least one project for the new account — you administer users per project",
)
outside = [p for p in requested if p not in managed]
if outside:
raise HTTPException(status_code=403, detail="You don't administer the users of one of those projects")
valid = set(db.scalars(select(models.Project.id).where(models.Project.id.in_(requested))).all()) if requested else set()
missing = [p for p in requested if p not in valid]
if missing:
raise HTTPException(status_code=400, detail="One of those projects no longer exists")
u = models.User(
id=gen_id("user"),
username=body.username.strip(),
email=body.email.strip(),
full_name=body.full_name.strip(),
role=body.role,
project_role=body.project_role.strip()[:120],
)
db.add(u)
# Flush the account before adding memberships that point at it. The ORM decides
# flush order from relationship() declarations, and models.py deliberately has
# none (plain columns + ForeignKey), so it will happily emit the project_members
# INSERT before the users one — which the database then rejects. Without this the
# whole call fails with a foreign-key violation on any engine that actually
# enforces them, which is every engine we run: Postgres always, and SQLite since
# db.py started setting `PRAGMA foreign_keys=ON`.
db.flush()
log_event(db, actor, "user_created", "user", u.id, summary=u.username,
detail={"role": u.role, "project_role": u.project_role,
"projects": len(valid)})
for pid in requested:
if pid in valid:
grant_project_access(db, u.id, pid)
if valid:
log_event(db, actor, "project_access_changed", "user", u.id, summary=u.username,
detail={"projects": len(valid), "reason": "created_with_access"})
db.commit()
db.refresh(u)
return directory_entry(db, u, actor)
@app.post("/api/auth/users/{user_id}/active")
def set_user_active(user_id: str, body: ActiveIn, actor: models.User = Depends(require_user_manager), db: Session = Depends(get_db)):
u = load_target_user(db, user_id)
require_see_user(db, actor, u)
require_manage_user(db, actor, u)
if u.id == actor.id and not body.is_active:
raise HTTPException(status_code=400, detail="You cannot disable your own account")
u.is_active = body.is_active
log_event(db, actor, "user_enabled" if body.is_active else "user_disabled", "user", u.id,
summary=u.username, detail={"is_active": bool(body.is_active)})
db.commit()
return directory_entry(db, u, actor)
@app.post("/api/auth/users/{user_id}/role")
def set_user_role(user_id: str, body: RoleIn, actor: models.User = Depends(require_user_manager), db: Session = Depends(get_db)):
"""Change a user's PERMISSIONS role. Their job function on the project is separate
— see set_user_project_role.
Guards: you can't change your own role (avoids self-lockout), the last remaining
admin can't be demoted (keeps the app manageable), and a super user may only hand
out the roles in `grantable_roles` — never admin or another super user."""
allowed = grantable_roles(actor)
if body.role not in allowed:
raise HTTPException(status_code=400, detail=f"role must be one of {', '.join(allowed)}")
u = load_target_user(db, user_id)
require_see_user(db, actor, u)
require_manage_user(db, actor, u)
if u.id == actor.id:
raise HTTPException(status_code=400, detail="You cannot change your own role")
if auth.is_admin(u) and body.role != auth.ROLE_ADMIN:
other_admins = db.scalars(
select(models.User.id).where(
(models.User.role == auth.ROLE_ADMIN)
& (models.User.id != u.id)
& (models.User.is_active.is_(True))
)
).all()
if not other_admins:
raise HTTPException(status_code=400, detail="Can't remove the last admin account")
old_role = u.role
u.role = body.role
detail = {"from": old_role, "to": body.role}
# Promoting someone to admin retires their default-member flag: an admin already
# reaches every project, so the flag would do nothing except sit there invisibly
# (the console shows admins no controls) and come back to life the day they are
# demoted. Same reasoning as clearing the role in set_user_auto_add.
if body.role == auth.ROLE_ADMIN and (u.auto_add_projects or u.auto_add_role):
u.auto_add_projects = False
u.auto_add_role = ""
detail["auto_add_cleared"] = True
log_event(db, actor, "role_changed", "user", u.id, summary=u.username, detail=detail)
db.commit()
db.refresh(u)
return directory_entry(db, u, actor)
@app.post("/api/auth/users/{user_id}/project-role")
def set_user_project_role(user_id: str, body: ProjectRoleIn, actor: models.User = Depends(require_user_manager), db: Session = Depends(get_db)):
"""Set a user's job function on the project (Project Manager, Superintendent,
…). Purely descriptive — it grants nothing. This is what the SOP team pickers
and notification routing read, so it's worth keeping accurate."""
u = load_target_user(db, user_id)
require_see_user(db, actor, u)
require_manage_user(db, actor, u)
old = u.project_role or ""
u.project_role = (body.project_role or "").strip()[:120]
log_event(db, actor, "project_role_changed", "user", u.id, summary=u.username,
detail={"from": old, "to": u.project_role})
db.commit()
db.refresh(u)
return directory_entry(db, u, actor)
@app.post("/api/auth/users/{user_id}/auto-add")
def set_user_auto_add(user_id: str, body: AutoAddIn, admin: models.User = Depends(auth.require_admin), db: Session = Depends(get_db)):
"""Flag a user as a default member of every project created from here on, with
an optional role on those projects. It only touches NEW projects — existing
assignments stay under the admin's hand (see set_user_projects), because
back-filling everyone onto historical jobs is never what this flag means.
App-admin only, unlike the rest of user administration: this is a standing rule
about every project that will ever exist, including the ones a super user has no
part in."""
allowed = ("",) + auth.PROJECT_SCOPED_ROLES
role = (body.role or "").strip()
if role not in allowed:
raise HTTPException(
status_code=400,
detail=f"role must be '' (inherit) or one of {', '.join(auth.PROJECT_SCOPED_ROLES)}",
)
u = db.get(models.User, user_id)
if not u:
raise HTTPException(status_code=404, detail="User not found")
u.auto_add_projects = bool(body.auto_add)
# A role left behind on a switched-off flag is a trap: it would quietly take
# effect the day someone switches the flag back on.
u.auto_add_role = role if u.auto_add_projects else ""
log_event(db, admin, "auto_add_changed", "user", u.id, summary=u.username,
detail={"auto_add": bool(u.auto_add_projects), "role": u.auto_add_role})
db.commit()
db.refresh(u)
return u.to_dict()
@app.delete("/api/auth/users/{user_id}")
def delete_user(user_id: str, actor: models.User = Depends(require_user_manager), db: Session = Depends(get_db)):
u = load_target_user(db, user_id)
require_see_user(db, actor, u)
require_manage_user(db, actor, u)
if u.id == actor.id:
raise HTTPException(status_code=400, detail="You cannot delete your own account")
log_event(db, actor, "user_deleted", "user", u.id, summary=u.username, detail={"role": u.role})
db.delete(u)
db.commit()
return {"deleted": user_id}
@app.get("/api/auth/users/{user_id}/projects")
def get_user_projects(user_id: str, actor: models.User = Depends(require_user_manager), db: Session = Depends(get_db)):
"""Which projects a user is assigned to, plus the project list to choose from.
(Admins implicitly access every project regardless of what's ticked here.)
A super user is shown ONLY the projects they administer, and `other_projects` says
how many more the person is on — enough for the dialog to be honest that it is
editing a slice of this account's access, without naming jobs that aren't theirs."""
u = load_target_user(db, user_id)
require_see_user(db, actor, u)
require_manage_user(db, actor, u)
rows = db.scalars(select(models.ProjectMember).where(models.ProjectMember.user_id == user_id)).all()
managed = managed_project_ids(db, actor)
if managed is None:
projects = db.scalars(select(models.Project).order_by(models.Project.name)).all()
in_scope = rows
else:
projects = db.scalars(
select(models.Project).where(models.Project.id.in_(managed)).order_by(models.Project.name)
).all() if managed else []
in_scope = [r for r in rows if r.project_id in managed]
return {
"user": directory_entry(db, u, actor),
"assigned": [r.project_id for r in in_scope],
# Per-project role overrides, keyed by project id ('' = inherit the account's).
"roles": {r.project_id: (r.role or "") for r in in_scope},
# Archived projects stay on this list on purpose — an existing assignment has
# to remain visible and removable — but they're flagged so the dialog can say
# so, rather than offering a finished job as though it were live work.
"projects": [{"id": p.id, "name": p.name, "number": p.number,
"archived": p.archived_at is not None} for p in projects],
"other_projects": len(rows) - len(in_scope),
"grantable_project_roles": list(
auth.PROJECT_SCOPED_ROLES if auth.is_admin(actor)
else (auth.ROLE_PROJECT_ADMIN, auth.ROLE_PROJECT_USER)
),
}
@app.put("/api/auth/users/{user_id}/projects")
def set_user_projects(user_id: str, body: ProjectAssignIn, actor: models.User = Depends(require_user_manager), db: Session = Depends(get_db)):
"""Replace a user's project assignments with the given set.
For an app admin the given set IS the whole answer. For a super user it replaces
only their own slice: memberships on projects they don't administer are left
exactly as they were, because a payload that simply omits them would otherwise cut
someone off from a job the caller can't even see."""
u = load_target_user(db, user_id)
require_see_user(db, actor, u)
require_manage_user(db, actor, u)
requested = [p for p in dict.fromkeys(body.project_ids) if p]
for pid in requested:
check_id(pid)
managed = managed_project_ids(db, actor)
if managed is not None:
outside = [p for p in requested if p not in managed]
if outside:
raise HTTPException(status_code=403, detail="You don't administer the users of one of those projects")
valid = set(db.scalars(select(models.Project.id).where(models.Project.id.in_(requested))).all()) if requested else set()
# A super user may hand out the project-scoped roles below their own; only an app
# admin can make someone a super user on a project. Anything unrecognised falls
# back to inheriting the account's own role.
allowed = auth.PROJECT_SCOPED_ROLES if auth.is_admin(actor) else (auth.ROLE_PROJECT_ADMIN, auth.ROLE_PROJECT_USER)
roles = {pid: r for pid, r in (body.roles or {}).items() if r in allowed}
# Rebuild only the rows this caller owns. Scoping the DELETE is the whole of the
# "leave other jobs alone" guarantee — get it wrong and a super user's save
# silently revokes access everywhere else.
doomed = delete(models.ProjectMember).where(models.ProjectMember.user_id == user_id)
if managed is not None:
# in_() on an empty set is a valid always-false predicate, so a caller who
# administers nothing deletes nothing (require_manage_user already refused them).
doomed = doomed.where(models.ProjectMember.project_id.in_(managed))
db.execute(doomed)
for pid in valid:
db.add(models.ProjectMember(id=gen_id("pm"), user_id=user_id, project_id=pid,
role=roles.get(pid, "")))
log_event(db, actor, "project_access_changed", "user", u.id, summary=u.username,
detail={"projects": len(valid),
"scope": "all" if managed is None else "managed",
"overrides": {p: r for p, r in roles.items() if p in valid}})
db.commit()
return {"assigned": sorted(valid), "roles": {p: roles.get(p, "") for p in sorted(valid)}}
# ── Projects ─────────────────────────────────────────────────────────────────
@app.post("/api/projects")
def upsert_project(body: ProjectIn, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
check_id(body.id)
proj = db.get(models.Project, body.id) if body.id else None
is_new = proj is None
if not is_new:
require_project_access(db, user, proj.id)
require_project_writable(db, user, proj.id, "Editing a project")
if proj is None:
proj = models.Project(id=body.id or gen_id("proj"))
db.add(proj)
proj.name = body.name
proj.number = body.number
proj.client = body.client
proj.division = body.division
proj.site = body.site
proj.sample = body.sample
proj.created_by = body.created_by or proj.created_by
proj.data = body.data
log_event(db, user, "created" if is_new else "updated", "project", proj.id,
project_id=proj.id, summary=(proj.name or proj.number or proj.id))
db.commit()
# A project created by a non-admin auto-grants its creator access. If that
# creator is ALSO a standing default member, this is the row that sticks —
# add_default_members below never overwrites an existing membership — so it has
# to carry the role they'd have been given, or someone whose flag says
# "Project Admin on every job" would silently land as a plain member on the one
# job they started themselves.
if is_new and not auth.is_admin(user):
creator_role = (user.auto_add_role or "").strip() if user.auto_add_projects else ""
grant_project_access(db, user.id, proj.id, creator_role)
db.commit()
# …and anyone flagged as a default member joins at creation time too.
if is_new:
add_default_members(db, proj.id, user)
db.commit()
db.refresh(proj)
return proj.to_dict()
@app.get("/api/projects")
def list_projects(
archived: str = Query("exclude", description="exclude (default) | only | all"),
user: models.User = Depends(auth.get_current_user),
db: Session = Depends(get_db),
):
stmt = scope_to_access(select(models.Project), models.Project.id, db, user)
if archived == "only":
stmt = stmt.where(models.Project.archived_at.is_not(None))
elif archived != "all":
stmt = stmt.where(models.Project.archived_at.is_(None)) # default: hide archived
rows = db.scalars(stmt.order_by(models.Project.updated_at.desc())).all()
# D7 / T9.8: archived projects are readable by PROJECT ADMINS only - anyone
# below that sees them nowhere, counts and pickers included. The default
# listing already excludes them; asking for them is what gets gated, and it
# is gated per project, so admin-on-Job-A does not surface archived Job B.
if archived != "exclude":
rows = [p for p in rows
if p.archived_at is None
or effective_role(db, user, p.id) in (
auth.ROLE_ADMIN, auth.ROLE_PROJECT_SUPER, auth.ROLE_PROJECT_ADMIN)]
return [p.summary() for p in rows]
@app.get("/api/projects/{project_id}")
def get_project(project_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
proj = db.get(models.Project, project_id)
if not proj:
raise HTTPException(status_code=404, detail="Project not found")
require_project_access(db, user, proj.id)
return proj.to_dict()
@app.delete("/api/projects/{project_id}")
def delete_project(project_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
proj = db.get(models.Project, project_id)
if not proj:
raise HTTPException(status_code=404, detail="Project not found")
# Cascades to every SOP and work package on the project — Project Admin only.
require_project_admin(db, user, proj.id, "Deleting a project")
log_event(db, user, "deleted", "project", proj.id, project_id=proj.id,
summary=(proj.name or proj.number or proj.id))
db.delete(proj)
db.commit()
return {"deleted": project_id}
@app.post("/api/projects/{project_id}/archive")
def archive_project(project_id: str, body: ArchiveIn, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
"""Archive (or unarchive) a whole project — it drops out of every picker,
switcher and search, and freezes read-only, without losing a thing. This is how
a finished job gets out of everyone's way while staying on the record; delete is
still there for a job that should never have existed."""
proj = db.get(models.Project, project_id)
if not proj:
raise HTTPException(status_code=404, detail="Project not found")
# Takes the whole job out of circulation for everyone on it, so it sits at the
# same bar as deleting it. Note the deliberate absence of require_project_writable
# — unarchiving is the one write an archived project must still accept.
require_project_admin(db, user, proj.id, "Archiving a project")
was_archived = proj.archived_at is not None
if body.archived and not was_archived:
proj.archived_at = models.utcnow()
log_event(db, user, "archived", "project", proj.id, project_id=proj.id,
summary=(proj.name or proj.number or proj.id))
elif not body.archived and was_archived:
proj.archived_at = None
log_event(db, user, "unarchived", "project", proj.id, project_id=proj.id,
summary=(proj.name or proj.number or proj.id))
db.commit()
db.refresh(proj)
return proj.to_dict()
# ── SOPs ─────────────────────────────────────────────────────────────────────
@app.post("/api/sops")
def upsert_sop(body: SopIn, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
check_id(body.id)
require_project_access(db, user, body.project_id)
# Both ends are checked: the project the SOP is being written INTO here, and the
# one it currently sits on below — so an archived job can be neither edited nor
# used as a source to move a SOP out of.
require_project_writable(db, user, body.project_id, "Saving a SOP")
sop = db.get(models.Sop, body.id) if body.id else None
if sop is not None:
require_project_access(db, user, sop.project_id)
require_project_writable(db, user, sop.project_id, "Saving a SOP")
# The SOP is the project's baseline: once it's been completed, changing it
# is a Project Admin action. Authoring and revising a draft is open to any
# project member, including marking it complete the first time.
if sop.complete:
require_project_admin(db, user, sop.project_id, "Changing a completed SOP")
is_new = sop is None
if sop is None:
sop = models.Sop(id=body.id or gen_id("sop"))
db.add(sop)
sop.project_id = body.project_id
sop.name = body.name
sop.number = body.number
sop.complete = body.complete
sop.created_by = body.created_by or sop.created_by
sop.data = body.data
log_event(db, user, "completed" if body.complete else ("created" if is_new else "updated"),
"sop", sop.id, project_id=sop.project_id, summary=(sop.name or sop.number or sop.id),
detail={"complete": bool(body.complete)})
db.commit()
db.refresh(sop)
return sop.to_dict()
@app.get("/api/sops")
def list_sops(project_id: Optional[str] = Query(None), full: bool = Query(False), user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
stmt = select(models.Sop)
if project_id:
stmt = stmt.where(models.Sop.project_id == project_id)
stmt = scope_to_access(stmt, models.Sop.project_id, db, user)
rows = db.scalars(stmt.order_by(models.Sop.updated_at.desc())).all()
# full=true includes the data JSON (the whole SOP document) for hydration;
# the default summary view stays lean for listing.
return [(s.to_dict() if full else s.summary()) for s in rows]
@app.get("/api/sops/latest")
def latest_sop(complete: Optional[bool] = None, project_id: Optional[str] = Query(None), user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
stmt = select(models.Sop)
if complete is not None:
stmt = stmt.where(models.Sop.complete == complete)
if project_id:
stmt = stmt.where(models.Sop.project_id == project_id)
stmt = scope_to_access(stmt, models.Sop.project_id, db, user)
sop = db.scalars(stmt.order_by(models.Sop.updated_at.desc()).limit(1)).first()
if not sop:
raise HTTPException(status_code=404, detail="No SOP found")
return sop.to_dict()
@app.get("/api/sops/{sop_id}")
def get_sop(sop_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
sop = db.get(models.Sop, sop_id)
if not sop:
raise HTTPException(status_code=404, detail="SOP not found")
require_project_access(db, user, sop.project_id)
return sop.to_dict()
@app.delete("/api/sops/{sop_id}")
def delete_sop(sop_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
sop = db.get(models.Sop, sop_id)
if not sop:
raise HTTPException(status_code=404, detail="SOP not found")
require_project_admin(db, user, sop.project_id, "Deleting a SOP")
require_project_writable(db, user, sop.project_id, "Deleting a SOP")
log_event(db, user, "deleted", "sop", sop.id, project_id=sop.project_id,
summary=(sop.name or sop.number or sop.id))
db.delete(sop)
db.commit()
return {"deleted": sop_id}
# ── Release gates (constraints + predecessors) ─────────────────────────────────
# Status ladder. The front end's STATUS_ORDER (wp-creation-app.js) has no "Issue"
# in it at all - the hold is a BRANCH off the released states, not a rung. It sits
# in this list so unknown statuses can still be told apart from known ones, but
# _released() must never count it: counting it is what let every transition out of
# hold skip the release gates as "already released", which made /status a side
# door past an open constraint (CR-015 / T7.3).
STATUS_ORDER = ["Draft", "Scheduled", "Issued", "In Progress", "Ready for QA", "Issue", "QC", "Closed"]
QA_READY_STATUS = "Ready for QA"
ISSUED_IDX = STATUS_ORDER.index("Issued")
DONE_STATUS = "Closed"
HOLD_STATUS = "Issue"
def _released(status: Optional[str]) -> bool:
"""Is this status a field state (Issued or beyond)? The hold is NOT one: a
package leaves hold through the gates, and enters it without them."""
if status == HOLD_STATUS:
return False
try:
return STATUS_ORDER.index(status or "") >= ISSUED_IDX
except ValueError:
return False
def predecessor_ids(data: Optional[dict]) -> list[str]:
"""Work-package ids this package waits on. Ignores anything malformed rather
than failing a save — a bad id simply can't block."""
raw = (data or {}).get("predecessors") or []
if not isinstance(raw, list):
return []
return [p for p in raw if isinstance(p, str) and _ID_RE.match(p)]
def gate_override(data: Optional[dict]) -> Optional[dict]:
"""A deliberate, reasoned override of the predecessor gate. Planners legitimately
need to stage packages ahead of the work finishing, so the gate is refusable —
but only explicitly, and it lands in the audit log."""
ov = (data or {}).get("gateOverride")
if isinstance(ov, dict) and str(ov.get("reason") or "").strip():
return ov
return None
def _urgent_constraint_override(data: Optional[dict], open_names: list) -> bool:
"""D4: only an URGENT package may release past an open constraint, and only
through the audited override - the same gateOverride record the predecessor
gate uses, never a separate path. The override must NAME every constraint it
covers: a constraint opened after the reason was written cannot ride through
on it. Normal and High are unchanged - a hard refusal."""
if str((data or {}).get("priority") or "") != "Urgent":
return False
ov = gate_override(data)
if not ov:
return False
covered = ov.get("constraints")
if not isinstance(covered, list):
return False
return {str(n) for n in open_names} <= {str(c) for c in covered}
def blocking_predecessors(db: Session, wp_id: Optional[str], data: Optional[dict]) -> list[dict]:
"""Predecessors that are not Closed yet. A predecessor that no longer exists is
NOT blocking — a deleted package must not freeze everything downstream."""
ids = [p for p in predecessor_ids(data) if p != wp_id]
if not ids:
return []
rows = db.scalars(select(models.WorkPackage).where(models.WorkPackage.id.in_(ids))).all()
return [
{"id": r.id, "number": r.number, "subject": r.subject, "status": r.status}
for r in rows if r.status != DONE_STATUS
]
def check_predecessor_cycle(db: Session, wp_id: str, data: Optional[dict]) -> None:
"""Refuse a predecessor set that would make A wait on itself (directly or
through a chain). Walks the graph from the proposed predecessors; the visited
set also bounds the walk, so a cycle that already exists elsewhere in the data
can't spin here."""
start = [p for p in predecessor_ids(data)]
if wp_id in start:
raise HTTPException(status_code=400, detail="A work package cannot be its own predecessor.")
seen, stack = set(), list(start)
while stack:
cur = stack.pop()
if cur in seen:
continue
seen.add(cur)
row = db.get(models.WorkPackage, cur)
if row is None:
continue
nxt = predecessor_ids(row.data)
if wp_id in nxt:
raise HTTPException(
status_code=400,
detail=f"That would create a circular dependency ({row.number or row.id} already waits on this package).",
)
stack.extend(n for n in nxt if n not in seen)
def enforce_release_gates(db: Session, wp_id: Optional[str], data: Optional[dict],
new_status: str, old_status: Optional[str]) -> None:
"""Refuse a move to Issued (or beyond) while a release gate is unmet. Applied to
every path that can set a status — the plain upsert included, since that's how
the browser and the offline outbox save."""
if not _released(new_status) or _released(old_status):
return # not a release transition
constraints = (data or {}).get("constraints") or []
open_names = [c.get("name") for c in constraints if isinstance(c, dict) and c.get("status") == "open"]
if open_names and not _urgent_constraint_override(data, open_names):
raise HTTPException(status_code=409, detail={
"message": "Open constraints block release", "open": open_names,
})
blockers = blocking_predecessors(db, wp_id, data)
if blockers and not gate_override(data):
raise HTTPException(status_code=409, detail={
"message": "Predecessors are not closed yet",
"blocking": [f"{b['number'] or b['id']} ({b['status']})" for b in blockers],
})
# ── Critical-constraint notification ───────────────────────────────────────────
# A constraint flagged CRITICAL on the SOP, reopened after the package was
# released, is announced by email. We detect it by comparing the incoming
# constraints against the stored ones during the normal upsert rather than adding a
# separate endpoint: the browser saves through the sync outbox, which only replays
# POST /api/wps, so anything hung off another route would be lost offline.
def reopened_critical(old_data: Optional[dict], new_data: Optional[dict]) -> list[str]:
old = {}
for c in (old_data or {}).get("constraints") or []:
if isinstance(c, dict) and c.get("name"):
old[c["name"]] = c.get("status")
out = []
for c in (new_data or {}).get("constraints") or []:
if not isinstance(c, dict) or not c.get("critical"):
continue
name = c.get("name")
if not name or c.get("status") != "open":
continue
if old.get(name) not in (None, "open"): # was cleared/na, now open
out.append(name)
return out
def project_sop_team(db: Session, project_id: Optional[str]) -> list[str]:
"""PM + CM user ids from the project's most recent complete SOP."""
if not project_id:
return []
sop = db.scalars(
select(models.Sop)
.where((models.Sop.project_id == project_id) & (models.Sop.complete.is_(True)))
.order_by(models.Sop.updated_at.desc())
.limit(1)
).first()
if not sop:
return []
data = sop.data or {}
# BL-021, fixed 2026-08-20: pushSOP writes the row as data={sop, state}, so
# the project block lives at data['sop']['project']. This read was one
# level too shallow - always {} - and the critical-reopen mail never
# reached the PM or CM its docstring promises. Same tolerant read as
# project_qa_group below: nested shape first, flat shape for hand-written rows.
proj = ((data.get("sop") or {}).get("project")
or data.get("project") or {})
return [i for i in (proj.get("pmId"), proj.get("cmId")) if i]
def project_qa_group(db: Session, project_id: Optional[str]) -> list["models.User"]:
"""D2: the QA group named on the project's latest complete SOP. pushSOP writes
the row as data={sop, state}, so the project block is data['sop']['project'] -
project_sop_team above read the flat shape until BL-021 was fixed
(2026-08-20); both now read nested-first, exactly alike."""
if not project_id:
return []
sop = db.scalars(
select(models.Sop)
.where((models.Sop.project_id == project_id) & (models.Sop.complete.is_(True)))
.order_by(models.Sop.updated_at.desc())
.limit(1)
).first()
if not sop:
return []
data = sop.data or {}
proj = ((data.get("sop") or {}).get("project")
or data.get("project") or {})
ids = [i for i in (proj.get("qaGroupIds") or []) if isinstance(i, str) and i]
if not ids:
return []
return list(db.scalars(select(models.User).where(models.User.id.in_(ids))))
def qa_rejection_comment(data: Optional[dict]) -> str:
rejs = [r for r in ((data or {}).get("qaRejections") or []) if isinstance(r, dict)]
if not rejs:
return ""
return str(rejs[-1].get("comment") or "").strip()
def enforce_qa_rejection_comment(data: Optional[dict], new_status: str,
old_status: Optional[str],
old_data: Optional[dict] = None) -> None:
"""CR-014: a rejection with no reason is the after-the-fact surprise this gate
exists to end. Runs before anything is written, like the release gates.
The comment must be FRESH: a rejection entry already on the record satisfied
the first version of this check, which made every rejection after the first
one free. New entry, non-empty comment, or the transition is refused."""
if old_status == QA_READY_STATUS and new_status == "In Progress":
new_rejs = [r for r in ((data or {}).get("qaRejections") or []) if isinstance(r, dict)]
old_rejs = [r for r in ((old_data or {}).get("qaRejections") or []) if isinstance(r, dict)]
fresh = len(new_rejs) > len(old_rejs) and str(new_rejs[-1].get("comment") or "").strip()
if not fresh:
raise HTTPException(status_code=409, detail={
"message": "Returning a package from Ready for QA requires a comment",
})
def qa_ready_body(user: "models.User", wp: "models.WorkPackage",
actor: "models.User", link: str) -> str:
# Number, title and location ride in the body — the 2026-08-20 decision
# restored the location the T7.6 done-when had excluded. The SCOPE summary
# stays out: scope text is document content, and the link is its summary.
who = actor.full_name or actor.username
name = user.full_name or user.username
return (
f"Hi {name},\n\n"
f"{who} moved {wp_titled(wp)} to Ready for QA.\n"
+ _where_line(wp) +
f"It is in the QA queue waiting to be accepted or returned.\n\n"
f"Open it here:\n{link}\n\n"
f"— This is an automated message from the Work Package Suite."
)
def qa_reject_body(user: "models.User", wp: "models.WorkPackage",
actor: "models.User", link: str) -> str:
who = actor.full_name or actor.username
name = user.full_name or user.username
return (
f"Hi {name},\n\n"
f"{who} returned {wp_titled(wp)} from Ready for QA to In Progress.\n"
+ _where_line(wp) +
f"The reason is recorded on the package.\n\n"
f"Open it here:\n{link}\n\n"
f"— This is an automated message from the Work Package Suite."
)
def notify_qa_transition(db: Session, wp: "models.WorkPackage",
actor: "models.User", rejected: bool) -> list:
"""Entering Ready for QA emails the QA group and nobody else (D2). A rejection
emails the owner AND the same group (D9's sibling decision). Deduplicated;
every send is an outbox row, so a failure is recorded, never silent."""
recipients = {u.id: u for u in project_qa_group(db, wp.project_id)}
if rejected and wp.assignee_id:
owner = db.get(models.User, wp.assignee_id)
if owner:
recipients[owner.id] = owner
out = []
link = wp_link(db, wp)
for u in recipients.values():
out.append(notify.enqueue(
db, user=u, kind="qa_rejected" if rejected else "qa_ready",
subject=(f"Returned from QA: {wp.number or 'work package'}" if rejected
else f"Ready for QA: {wp.number or 'work package'}"),
body=(qa_reject_body(u, wp, actor, link) if rejected
else qa_ready_body(u, wp, actor, link)),
link=link, wp_id=wp.id, project_id=wp.project_id,
))
return out
def hold_body(user: "models.User", wp: "models.WorkPackage", names: list[str],
actor: "models.User", link: str) -> str:
# Constraint names, number, title and location — customer context per the
# 2026-08-20 decision. No package contents; the link carries those.
who = user.full_name or user.username
by = actor.full_name or actor.username
which = ", ".join(names)
return (
f"Hi {who},\n\n"
f"A critical constraint was reopened on {wp_titled(wp)} "
f"after it was released to the field, so the package is on hold.\n\n"
f"Constraint: {which}\n"
+ _where_line(wp) +
f"Reopened by: {by}\n\n"
f"Open the package:\n{link}\n\n"
f"— This is an automated message from the Work Package Suite."
)
def kitting_body(user: "models.User", wp: "models.WorkPackage", actor: "models.User",
old_status: str, new_status: str, link: str) -> str:
# Same convention as the assignment and hold mails: a greeting, one line of
# what happened, the deep link, the footer. The delivery location is MIMO
# logistics (where material stages), not package contents - the wave 8 spec
# names it in the done-when. deliveryLoc arrives with CR-012 (T8.4);
# mimoLoc is what exists today, so both are read.
who = actor.full_name or actor.username
name = user.full_name or user.username
delivery = str((wp.data or {}).get("deliveryLoc")
or (wp.data or {}).get("mimoLoc") or "").strip() or "not set"
return (
f"Hi {name},\n\n"
f"{who} moved kitting on {wp_titled(wp)} from "
f"{old_status or 'Not Started'} to {new_status or 'Not Started'}.\n"
f"Delivery location: {delivery}.\n\n"
f"Open it here:\n{link}\n\n"
f"— This is an automated message from the Work Package Suite."
)
def material_request_body(user: "models.User", wp: "models.WorkPackage",
actor: "models.User", n_lines: int, needed: str,
delivery: str, link: str) -> str:
who = actor.full_name or actor.username
name = user.full_name or user.username
needed_line = f" needed by {needed}" if needed else ""
return (
f"Hi {name},\n\n"
f"{who} raised a material request on {wp_titled(wp)}: "
f"{n_lines} line{'' if n_lines == 1 else 's'}{needed_line}.\n"
f"Delivery location: {delivery}.\n\n"
f"Open it here:\n{link}\n\n"
f"— This is an automated message from the Work Package Suite."
)
def notify_kitting_change(db: Session, wp: "models.WorkPackage", actor: "models.User",
old_status: str, new_status: str) -> list:
"""CR-011: the package's distribution list plus its warehouse owner (CR-010's
default recipient), minus the actor, deduplicated. COALESCED: if an unsent
kitting notification already exists for this package and recipient, it is
rewritten to the newest transition instead of joined by a sibling - rapid
consecutive changes produce one email saying where kitting ended up, not a
burst of near-identical ones."""
ids = [i for i in ((wp.data or {}).get("distributionIds") or []) if isinstance(i, str)]
ko = (wp.data or {}).get("kitOwnerId")
if isinstance(ko, str) and ko:
ids.append(ko)
link = wp_link(db, wp)
subject = f"Kitting {new_status or 'updated'}: {wp.number or 'work package'}"
existing = {n.user_id: n for n in db.scalars(
select(models.Notification).where(
(models.Notification.wp_id == wp.id)
& (models.Notification.kind == "kitting_status")
& (models.Notification.status.in_(("pending", "skipped"))))).all()}
s_cfg = notify.get_settings(db)
deliverable = bool(s_cfg.get("email_enabled")) and notify.smtp_ready(s_cfg)
seen, out = set(), []
for uid in ids:
if uid in seen or uid == actor.id:
continue
seen.add(uid)
u = db.get(models.User, uid)
if not u or not u.is_active:
continue
held = existing.get(uid)
if held is not None:
held.subject = subject[:300]
held.body = kitting_body(u, wp, actor, old_status, new_status, link)
held.link = link[:500]
# A row held while email was OFF stays 'skipped' forever unless the
# change that finds email ON promotes it - otherwise turning the
# gate on silently orphans everything coalesced before it.
if deliverable and u.email and held.status == "skipped":
held.status = "pending"
out.append(held)
continue
out.append(notify.enqueue(
db, user=u, kind="kitting_status", subject=subject,
body=kitting_body(u, wp, actor, old_status, new_status, link),
link=link, wp_id=wp.id, project_id=wp.project_id,
))
return out
def notify_critical_reopen(db: Session, wp: "models.WorkPackage", names: list[str],
actor: "models.User") -> list["models.Notification"]:
"""Owner + PM + CM + everyone on the package's distribution list, minus whoever
did it (they already know) and minus duplicates."""
ids = []
if wp.assignee_id:
ids.append(wp.assignee_id)
ids.extend(project_sop_team(db, wp.project_id))
ids.extend([i for i in ((wp.data or {}).get("distributionIds") or []) if isinstance(i, str)])
seen, out = set(), []
link = wp_link(db, wp)
for uid in ids:
if uid in seen or uid == actor.id:
continue
seen.add(uid)
u = db.get(models.User, uid)
if not u or not u.is_active:
continue
out.append(notify.enqueue(
db, user=u, kind="wp_constraint_reopened",
subject=f"On hold: {wp.number or 'work package'} — critical constraint reopened",
body=hold_body(u, wp, names, actor, link),
link=link, wp_id=wp.id, project_id=wp.project_id,
))
return out
# ── Work Packages ────────────────────────────────────────────────────────────
@app.post("/api/wps")
def upsert_wp(body: WpIn, background_tasks: BackgroundTasks, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
check_id(body.id)
check_id(body.parent_id)
check_id(body.assignee_id)
require_project_access(db, user, body.project_id)
# Destination and origin both have to be live — see upsert_sop.
require_project_writable(db, user, body.project_id, "Saving a work package")
wp = db.get(models.WorkPackage, body.id) if body.id else None
if wp is not None:
require_project_access(db, user, wp.project_id)
require_project_writable(db, user, wp.project_id, "Saving a work package")
is_new = wp is None
old_status = None if is_new else wp.status
old_assignee = None if is_new else wp.assignee_id
old_data = None if is_new else (wp.data or {})
if wp is None:
wp = models.WorkPackage(id=body.id or gen_id("wp"))
db.add(wp)
# Predecessors: reject a cycle, and refuse a release while a gate is unmet.
# Both run before anything is written so a rejected save changes nothing.
wp_id_for_checks = body.id or wp.id
check_predecessor_cycle(db, wp_id_for_checks, body.data)
enforce_release_gates(db, wp_id_for_checks, body.data, body.status, old_status)
enforce_qa_rejection_comment(body.data, body.status, old_status, old_data)
wp.project_id = body.project_id
wp.sop_id = body.sop_id
wp.parent_id = body.parent_id
wp.number = body.number
wp.subject = body.subject
wp.type = body.type
wp.status = body.status
new_assignee = body.assignee_id or None
if new_assignee:
require_assignable(db, new_assignee, body.project_id)
wp.assignee_id = new_assignee
wp.created_by = body.created_by or wp.created_by
# data["files"] is SERVER-owned (CR-007): it mirrors the wp_files table and
# is rewritten by the upload/delete/patch routes. A client that saved before
# an upload landed would otherwise erase the list with its stale copy.
if not is_new:
_stored_files = (old_data or {}).get("files")
if _stored_files is not None:
body.data = dict(body.data or {})
body.data["files"] = _stored_files
wp.data = body.data
if is_new:
_act, _detail = "created", {"status": wp.status}
elif old_status != wp.status:
_act, _detail = "status_changed", {"from": old_status, "to": wp.status}
else:
_act, _detail = "updated", {"status": wp.status}
log_event(db, user, _act, "wp", wp.id, project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id), detail=_detail)
notifs = []
# A reasoned override of the predecessor gate is worth its own audit line —
# "released early, and here's why" is exactly what a reviewer looks for later.
ov = gate_override(body.data)
if ov and _released(wp.status) and not _released(old_status):
blockers = blocking_predecessors(db, wp.id, body.data)
_ov_constraints = [str(c) for c in ov.get("constraints") or [] if c]
log_event(db, user, "gate_overridden", "wp", wp.id, project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id),
detail={"reason": str(ov.get("reason"))[:300],
"blocking": [b["number"] or b["id"] for b in blockers],
"constraints": _ov_constraints})
# CR-015: holds and releases are history, not just state. Entering or leaving
# the hold branch gets its own audit line with the actor and the reason the
# browser recorded in data.holds - `status_changed` alone says from/to but
# not why, and the WHY is what a notice of delay is built from.
if not is_new and old_status != wp.status and "Issue" in (old_status, wp.status):
_hlds = [h for h in ((wp.data or {}).get("holds") or []) if isinstance(h, dict)]
if wp.status == "Issue":
_h = next((h for h in reversed(_hlds) if not h.get("released")), None)
log_event(db, user, "hold_logged", "wp", wp.id, project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id),
detail={"from": old_status,
"constraint": str((_h or {}).get("constraint") or "")[:200],
"reason": str((_h or {}).get("details") or "")[:300]})
else:
_h = next((h for h in reversed(_hlds) if h.get("released")), None)
log_event(db, user, "hold_released", "wp", wp.id, project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id),
detail={"to": wp.status,
"constraint": str((_h or {}).get("constraint") or "")[:200],
"reason": str((_h or {}).get("details") or "")[:300]})
# A critical constraint reopened after release: log it and tell the people who
# need to know (owner, PM, CM, distribution).
if not is_new and (_released(old_status) or old_status == HOLD_STATUS):
reopened = reopened_critical(old_data, wp.data)
if reopened:
log_event(db, user, "constraint_reopened", "wp", wp.id, project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id),
detail={"critical": reopened, "status": wp.status})
notifs.extend(notify_critical_reopen(db, wp, reopened, user))
# CR-014: the QA gate. Entering Ready for QA tells the QA group their queue
# grew; a rejection tells the owner and the same group. Both write their own
# audit line - status_changed says from/to, these say what it MEANS.
if not is_new and old_status != wp.status:
if wp.status == QA_READY_STATUS:
log_event(db, user, "qa_ready", "wp", wp.id, project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id), detail={"from": old_status})
notifs.extend(notify_qa_transition(db, wp, user, rejected=False))
elif old_status == QA_READY_STATUS and wp.status == "In Progress":
log_event(db, user, "qa_rejected", "wp", wp.id, project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id),
detail={"comment": qa_rejection_comment(wp.data)[:300]})
notifs.extend(notify_qa_transition(db, wp, user, rejected=True))
# CR-011: a kitting status change tells the distribution list and the
# warehouse owner where the material stands. Detected here because the
# browser saves kitting through this upsert (the outbox replays it too).
if not is_new:
_kit_old = str((old_data or {}).get("kitStatus") or "")
_kit_new = str((wp.data or {}).get("kitStatus") or "")
if _kit_old != _kit_new:
log_event(db, user, "kitting_status_changed", "wp", wp.id,
project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id),
detail={"from": _kit_old, "to": _kit_new})
notifs.extend(notify_kitting_change(db, wp, user, _kit_old, _kit_new))
# CR-013 / T8.5: a new material request notifies the warehouse owner named
# on the package (CR-010) - the routing that replaces the informal funnel
# through one person. Same gate, same outbox, same link discipline.
if not is_new:
_mr_old = [r for r in ((old_data or {}).get("materialRequests") or []) if isinstance(r, dict)]
_mr_new = [r for r in ((wp.data or {}).get("materialRequests") or []) if isinstance(r, dict)]
if len(_mr_new) > len(_mr_old):
fresh = _mr_new[len(_mr_old):]
for req in fresh:
log_event(db, user, "material_requested", "wp", wp.id,
project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id),
detail={"lines": len(req.get("items") or []),
"neededBy": str(req.get("neededBy") or "")[:40]})
ko = (wp.data or {}).get("kitOwnerId")
owner = db.get(models.User, ko) if isinstance(ko, str) and ko else None
if owner and owner.is_active and owner.id != user.id:
link = wp_link(db, wp)
n_lines = sum(len(r.get("items") or []) for r in fresh)
needed = str(fresh[-1].get("neededBy") or "").strip()
deliv = str(fresh[-1].get("deliveryLoc") or "").strip() or "not set"
notifs.append(notify.enqueue(
db, user=owner, kind="material_requested",
subject=f"Material request: {wp.number or 'work package'}",
body=material_request_body(owner, wp, user, n_lines, needed, deliv, link),
link=link, wp_id=wp.id, project_id=wp.project_id,
))
# Notify a newly-assigned owner (skip self-assignment).
notif = None
if new_assignee and new_assignee != old_assignee and new_assignee != user.id:
assignee = db.get(models.User, new_assignee)
if assignee:
link = wp_link(db, wp)
notif = notify.enqueue(
db, user=assignee, kind="wp_assigned",
subject=f"You were assigned {wp.number or 'a work package'}",
body=assign_body(assignee, wp, user, link),
link=link, wp_id=wp.id, project_id=wp.project_id,
)
log_event(db, user, "assigned", "wp", wp.id, project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id), detail={"to": assignee.username})
db.commit()
db.refresh(wp)
if notif is not None:
notifs.append(notif)
for n in notifs:
background_tasks.add_task(notify.deliver, n.id)
return wp.to_dict()
@app.get("/api/wps")
def list_wps(
response: Response,
project_id: Optional[str] = Query(None),
sop_id: Optional[str] = Query(None),
parent_id: Optional[str] = Query(None),
status: Optional[str] = Query(None),
q: Optional[str] = Query(None, description="search number / subject / type"),
archived: str = Query("exclude", description="exclude (default) | only | all"),
limit: Optional[int] = Query(None, ge=1, le=1000),
offset: int = Query(0, ge=0),
full: bool = Query(False),
user: models.User = Depends(auth.get_current_user),
db: Session = Depends(get_db),
):
stmt = select(models.WorkPackage)
if project_id:
stmt = stmt.where(models.WorkPackage.project_id == project_id)
if sop_id:
stmt = stmt.where(models.WorkPackage.sop_id == sop_id)
if parent_id:
stmt = stmt.where(models.WorkPackage.parent_id == parent_id)
if status:
stmt = stmt.where(models.WorkPackage.status == status)
if archived == "only":
stmt = stmt.where(models.WorkPackage.archived_at.is_not(None))
elif archived != "all":
stmt = stmt.where(models.WorkPackage.archived_at.is_(None)) # default: hide archived
if q and q.strip():
like = f"%{q.strip()}%"
stmt = stmt.where(
models.WorkPackage.number.ilike(like)
| models.WorkPackage.subject.ilike(like)
| models.WorkPackage.type.ilike(like)
)
stmt = scope_to_access(stmt, models.WorkPackage.project_id, db, user)
# Report the pre-pagination total so the client can build a pager.
total = db.scalar(select(func.count()).select_from(stmt.subquery()))
response.headers["X-Total-Count"] = str(total or 0)
stmt = stmt.order_by(models.WorkPackage.updated_at.desc()).offset(offset)
if limit is not None:
stmt = stmt.limit(limit)
rows = db.scalars(stmt).all()
# full=true includes the data JSON (full package document) so the creator can
# rehydrate everything in one request; default stays lean for listing.
return [(w.to_dict() if full else w.summary()) for w in rows]
# Weighted completion by status, 0..1, so progress reads as a slope rather than
# done/not-done. Mirrors PROGRESS_W in wp-creation-app.js — the dashboard used to
# compute this in the browser from localStorage, which is exactly what B4 removes.
PROGRESS_WEIGHT = {
"Draft": 0.0, "Scheduled": 0.25, "Issue": 0.4, "Issued": 0.5,
"In Progress": 0.75, "QC": 0.9, "Closed": 1.0,
}
# The dimensions a location rollup is grouped by. Today a work package carries one
# free-text `location` ("building / level / sector / room"), so that is the single
# dimension. CR-004 replaces it with structured building / floor / sector in wave 6;
# when it does, only this tuple and the keys inside each group change — the response
# shape does not, which is what T4.1 means by "can be grouped by location without a
# schema change". CR-018's rollup consumes `by_location.groups` either way.
# CR-004 landed at T6.3, so this is now what it was designed to become. T4.1's
# note said "only this tuple and the keys inside each group change — the response
# shape does not", and that held: CR-018's rollup consumes `by_location.groups`
# exactly as it did when there was one free-text dimension.
LOCATION_DIMENSIONS = ("building", "floor", "sector")
# A package with no location is not dropped from the rollup. `CR-018`'s own
# done-when says so, and the reason is arithmetic: a group set that silently
# omits the unlocated packages does not add up to the project total, and a rollup
# that does not reconcile is worse than no rollup.
LOCATION_UNSET = "(unassigned)"
def _location_key(data: dict) -> dict:
"""The location dimensions of one package, as a dict keyed by dimension name.
Structured fields win; the free-text `location` a package captured before
CR-004 is kept as its own dimension value so those packages group together
under what they actually said rather than all collapsing into one bucket."""
structured = {d: (data.get(d) or "").strip() for d in LOCATION_DIMENSIONS}
if any(structured.values()):
return {k: v or LOCATION_UNSET for k, v in structured.items()}
legacy = (data.get("location") or "").strip()
if legacy:
# One dimension deep, deliberately: free text is not a hierarchy and
# pretending it is would put "FAB / LVL 1" under a building called
# "FAB / LVL 1".
return {"building": legacy, "floor": LOCATION_UNSET, "sector": LOCATION_UNSET}
return {d: LOCATION_UNSET for d in LOCATION_DIMENSIONS}
def _location_levels(loc_groups: dict) -> dict:
"""Totals at each level of the hierarchy, not only at the leaf.
`by_location.groups` is one row per distinct (building, floor, sector). The
question CR-018 asks — "what is on floor 2" — is a level above that, and
every level has to reconcile against the project total or the rollup is
decoration. Each level therefore sums EVERY package, including the ones whose
value at that level is unassigned."""
out: dict[str, list] = {}
for dim in LOCATION_DIMENSIONS:
buckets: dict[str, dict] = {}
for g in loc_groups.values():
# Grouped by the value AT THIS LEVEL alone, not by the tuple of levels
# above it. Each stored value is already a full path — a floor is
# `B-ONE/L1`, not `L1` — so it carries its own ancestry and is unique
# across buildings without being re-qualified. Only the unassigned
# bucket is shared, which is what it should be: "these have no floor
# recorded" is one answer, not one answer per building.
path = g["key"].get(dim, LOCATION_UNSET) or LOCATION_UNSET
slot = buckets.setdefault(path, {
"dimension": dim,
"path": path,
"total": 0, "release_ready": 0, "on_hold": 0, "overdue": 0,
"est_hours": 0.0, "actual_hours": 0.0,
})
for f in ("total", "release_ready", "on_hold", "overdue", "est_hours", "actual_hours"):
slot[f] += g[f]
out[dim] = [
{**b, "est_hours": round(b["est_hours"]), "actual_hours": round(b["actual_hours"])}
for b in sorted(buckets.values(), key=lambda b: b["path"])
]
return out
@app.get("/api/wps/metrics")
def wp_metrics(project_id: Optional[str] = Query(None), sop_id: Optional[str] = Query(None), user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
"""Every count and rollup the creator's dashboard shows, computed from the
database rather than from the caller's own browser (B4).
Masters (data.split == true) are excluded so a split package's hours are not
double-counted with its instances. Archived packages are excluded throughout.
Two people on the same project get the same numbers from this endpoint. They
did not when each browser derived them from its own localStorage, and neither
was told — which is the failure B4 exists to remove."""
stmt = select(models.WorkPackage).where(models.WorkPackage.archived_at.is_(None))
if project_id:
stmt = stmt.where(models.WorkPackage.project_id == project_id)
if sop_id:
stmt = stmt.where(models.WorkPackage.sop_id == sop_id)
stmt = scope_to_access(stmt, models.WorkPackage.project_id, db, user)
rows = db.scalars(stmt).all()
today = models.utcnow().date().isoformat()
by_status: dict[str, int] = {}
by_discipline: dict[str, int] = {}
loc_groups: dict[tuple, dict] = {}
prog_by_disc: dict[str, dict] = {}
gating: list[dict] = []
total = ready = on_hold = overdue = mine = 0
est_hours = actual_hours = 0.0
progress_sum = 0.0
for w in rows:
data = w.data or {}
if data.get("split"):
continue
total += 1
by_status[w.status] = by_status.get(w.status, 0) + 1
if w.status == "Issue":
on_hold += 1
if w.assignee_id and w.assignee_id == user.id:
mine += 1
due = (data.get("due") or "").strip()
is_overdue = bool(due and w.status != "Closed" and due < today)
if is_overdue:
overdue += 1
constraints = data.get("constraints") or []
open_constraints = [c for c in constraints if c.get("status") == "open"]
# `waitingOn` are predecessor packages not yet closed; the browser counted
# them as blocking too, so the server has to, or "release-ready" changes
# meaning the moment the dashboard stops computing it locally.
waiting_on = [x for x in (data.get("waitingOn") or []) if x]
blocked = bool(open_constraints or waiting_on)
if not blocked and w.status not in ("Closed", "Issue"):
ready += 1
if open_constraints:
gating.append({
"id": w.id, "number": w.number, "subject": w.subject,
"blocked_by": [
{"name": c.get("name") or "", "comment": c.get("comment") or ""}
for c in open_constraints
],
})
try:
est_hours += float(data.get("hours") or 0)
except (TypeError, ValueError):
pass
try:
actual_hours += float(data.get("actualHrs") or 0)
except (TypeError, ValueError):
pass
weight = PROGRESS_WEIGHT.get(w.status, 0.0)
progress_sum += weight
disciplines = data.get("disciplines") or ["(none)"]
for d in disciplines:
by_discipline[d] = by_discipline.get(d, 0) + 1
slot = prog_by_disc.setdefault(d, {"total": 0, "done": 0, "weight": 0.0})
slot["total"] += 1
slot["weight"] += weight
if w.status == "Closed":
slot["done"] += 1
key = _location_key(data)
kt = tuple(key.get(d, "(unset)") for d in LOCATION_DIMENSIONS) if len(key) == len(LOCATION_DIMENSIONS) else tuple(sorted(key.items()))
slot = loc_groups.setdefault(kt, {"key": key, "total": 0, "release_ready": 0,
"on_hold": 0, "overdue": 0, "by_status": {},
"est_hours": 0.0, "actual_hours": 0.0})
slot["total"] += 1
slot["by_status"][w.status] = slot["by_status"].get(w.status, 0) + 1
if not blocked and w.status not in ("Closed", "Issue"):
slot["release_ready"] += 1
if w.status == "Issue":
slot["on_hold"] += 1
if is_overdue:
slot["overdue"] += 1
# CR-018: hours roll up along the same dimensions. Actual Hours is the one
# CR-017 retained and is the reason that decision mattered — it is what
# makes a floor's real cost visible.
try:
slot["est_hours"] += float(data.get("hours") or 0)
except (TypeError, ValueError):
pass
try:
slot["actual_hours"] += float(data.get("actualHrs") or 0)
except (TypeError, ValueError):
pass
dimensions = list(LOCATION_DIMENSIONS)
if loc_groups:
first = next(iter(loc_groups.values()))["key"]
dimensions = list(first.keys())
return {
"total": total, "mine": mine, "release_ready": ready, "on_hold": on_hold,
"overdue": overdue,
"est_hours": round(est_hours), "actual_hours": round(actual_hours),
"by_status": by_status, "by_discipline": by_discipline,
"progress": {
"overall_pct": round(progress_sum / total * 100) if total else 0,
"by_discipline": [
{"name": d, "pct": round(v["weight"] / v["total"] * 100) if v["total"] else 0,
"done": v["done"], "total": v["total"]}
for d, v in sorted(prog_by_disc.items())
],
},
"gating": sorted(gating, key=lambda g: (g["number"] or "", g["id"])),
"by_location": {
"dimensions": dimensions,
"unassigned_key": LOCATION_UNSET,
"groups": [
{**g, "est_hours": round(g["est_hours"]), "actual_hours": round(g["actual_hours"])}
for g in sorted(loc_groups.values(),
key=lambda g: tuple(str(v) for v in g["key"].values()))
],
# Rolled up one level at a time as well as by the full triple, because
# "how many on floor 2" is the question CR-018 is actually about and
# summing the leaf groups in the browser would be the same per-browser
# arithmetic B4 removed.
"levels": _location_levels(loc_groups),
},
"generated_at": models.utcnow().isoformat(),
}
# ── Location taxonomy — CR-005 ────────────────────────────────────────────────
# Building / Floor / Sector, configured once per project rather than hard-coded.
# See models.LocationNode for why this stores codes as well as names, and why
# nothing here deletes.
LOCATION_LEVELS = ("building", "floor", "sector")
_SLUG_STRIP = re.compile(r"[^A-Z0-9]+")
def location_slug(name: str) -> str:
"""A stable code from a display name: upper-cased, non-alphanumerics collapsed
to a single dash, trimmed. `Level 2` -> `LEVEL-2`, `1P (chase)` -> `1P-CHASE`.
Derived ONCE, at import, and never recomputed — see LocationNode. Returns ''
when there is nothing to make a code from, which the caller turns into a
rejected row with a reason rather than a silently-skipped one."""
return _SLUG_STRIP.sub("-", (name or "").strip().upper()).strip("-")[:60]
def parse_location_rows(text: str) -> tuple[list[tuple[int, list[str]]], list[dict]]:
"""Split pasted or uploaded text into (rows, rejected).
Rows come back paired with their SOURCE line number, not their index among the
accepted ones. "Duplicate on row 12" has to mean row 12 of the file somebody is
looking at, or the report sends them to the wrong line.
Accepts comma, tab or semicolon separators — a paste out of Excel is
tab-separated and a saved CSV is not, and asking which one somebody has is a
question the machine can answer. A header line naming the levels is skipped.
Every rejection carries the line number and a reason. A silent skip is the
failure mode this endpoint exists to avoid: an import that says "42 rows" over
a file with 50 in it has lost eight and told nobody."""
rows: list[tuple[int, list[str]]] = []
rejected: list[dict] = []
for i, raw in enumerate((text or "").splitlines(), start=1):
line = raw.strip()
if not line:
continue
if "\t" in raw:
parts = [c.strip() for c in raw.split("\t")]
elif ";" in line and "," not in line:
parts = [c.strip() for c in line.split(";")]
else:
parts = [c.strip() for c in line.split(",")]
parts = [p.strip().strip('"').strip() for p in parts]
while parts and not parts[-1]:
parts.pop()
if not parts:
continue
low = [p.lower() for p in parts]
if i == 1 and low[:1] in (["building"], ["bldg"]):
continue # header row
if len(parts) > len(LOCATION_LEVELS):
rejected.append({"line": i, "text": line,
"reason": "more than 3 columns — expected building, floor, sector"})
continue
if not parts[0]:
rejected.append({"line": i, "text": line,
"reason": "no building — a floor or sector needs one above it"})
continue
# A gap in the middle ("B100,,1P") would attach a sector to nothing.
gap = next((n for n, p in enumerate(parts) if not p), None)
if gap is not None and any(parts[gap + 1:]):
rejected.append({"line": i, "text": line,
"reason": "a %s is named with no %s above it"
% (LOCATION_LEVELS[len(parts) - 1], LOCATION_LEVELS[gap])})
continue
parts = [p for p in parts if p]
if any(not location_slug(p) for p in parts):
rejected.append({"line": i, "text": line,
"reason": "no letters or digits to make a code from"})
continue
rows.append((i, parts))
return rows, rejected
def _location_tree(db: Session, project_id: str, include_inactive: bool):
stmt = select(models.LocationNode).where(models.LocationNode.project_id == project_id)
if not include_inactive:
stmt = stmt.where(models.LocationNode.active.is_(True))
nodes = db.scalars(stmt.order_by(models.LocationNode.path)).all()
return nodes
@app.get("/api/projects/{project_id}/locations")
def list_locations(project_id: str,
include_inactive: bool = Query(False),
user: models.User = Depends(auth.get_current_user),
db: Session = Depends(get_db)):
"""The project's taxonomy, flat, ordered by path so a caller can rebuild the
tree without a second query.
`include_inactive` defaults to FALSE, which is what makes a deactivated value
disappear from new work packages. It is available as true because a package
that already references a deactivated value still has to render its label —
deactivating hides a choice, it does not rewrite history."""
if not db.get(models.Project, project_id):
raise HTTPException(status_code=404, detail="Project not found")
require_project_access(db, user, project_id)
nodes = _location_tree(db, project_id, include_inactive)
return {
"project_id": project_id,
"levels": list(LOCATION_LEVELS),
"nodes": [n.to_dict() for n in nodes],
"counts": {lvl: sum(1 for n in nodes if n.level == lvl) for lvl in LOCATION_LEVELS},
"generated_at": models.utcnow().isoformat(),
}
class LocationImportIn(BaseModel):
text: str = ""
dry_run: bool = False
@app.post("/api/projects/{project_id}/locations/import")
def import_locations(project_id: str, body: LocationImportIn,
user: models.User = Depends(auth.get_current_user),
db: Session = Depends(get_db)):
"""Bulk import from CSV or a paste. Same code path for both: a file is read in
the browser and posted as text, because two parsers would be two sets of rules
about what a blank column means.
Reports rather than merges. A row naming a building/floor/sector combination
that already exists — in this file or in the project — comes back in
`duplicates` with its line number. Reusing a PARENT is not a duplicate:
`B100,L2,1P` and `B100,L2,2P` share a building and a floor by design, and it
is only the full path repeating that is a duplicate.
`dry_run` parses and reports without writing, which is what lets the wizard
show what an import will do before it does it."""
if not db.get(models.Project, project_id):
raise HTTPException(status_code=404, detail="Project not found")
# Any project member, matching how the SOP baseline itself is authored — the
# Project Admin gate is on CHANGING a completed SOP, not on writing one.
require_project_access(db, user, project_id)
require_project_writable(db, user, project_id, "The location list cannot be changed")
rows, rejected = parse_location_rows(body.text)
existing = {n.path: n for n in db.scalars(
select(models.LocationNode).where(models.LocationNode.project_id == project_id)
).all()}
before = set(existing)
created: list[dict] = []
duplicates: list[dict] = []
reactivated: list[dict] = []
seen_in_file: dict[str, int] = {}
next_sort = max((n.sort for n in existing.values()), default=0)
for line_no, parts in rows:
segs = [location_slug(p) for p in parts]
full = "/".join(segs)
if full in seen_in_file:
duplicates.append({"line": line_no, "path": full, "names": parts,
"reason": "already on line %d of this import" % seen_in_file[full]})
continue
seen_in_file[full] = line_no
if full in before:
node = existing[full]
if not node.active:
# Re-importing a value somebody deactivated is a request to bring it
# back, not a duplicate — and it must reuse the SAME row, or every
# work package pointing at the old path is orphaned.
if not body.dry_run:
node.active = True
reactivated.append({"path": full, "names": parts})
else:
duplicates.append({"line": line_no, "path": full, "names": parts,
"reason": "already in this project"})
continue
# Create any missing ancestors, then the leaf. Sharing a parent is the
# normal case, not a collision.
parent_id = None
for depth, seg in enumerate(segs):
sub = "/".join(segs[:depth + 1])
node = existing.get(sub)
if node is None:
next_sort += 1
node = models.LocationNode(
id=gen_id("loc"), project_id=project_id, parent_id=parent_id,
level=LOCATION_LEVELS[depth], code=seg, path=sub,
name=parts[depth], active=True, sort=next_sort,
created_by=user.username,
)
existing[sub] = node
if not body.dry_run:
db.add(node)
if sub not in before:
created.append({"path": sub, "level": LOCATION_LEVELS[depth],
"code": seg, "name": parts[depth]})
parent_id = node.id
result = {
"project_id": project_id, "dry_run": bool(body.dry_run),
"read": len(rows) + len(rejected),
"created": created, "duplicates": duplicates,
"reactivated": reactivated, "rejected": rejected,
}
if body.dry_run:
db.rollback()
return result
log_event(db, user, "locations_imported", "project", project_id, project_id,
summary="%d location value(s) added" % len(created),
detail={"created": len(created), "duplicates": len(duplicates),
"rejected": len(rejected), "reactivated": len(reactivated)})
db.commit()
return result
class MaterialIn(BaseModel):
description: str = ""
unit: str = ""
code: str = ""
class MaterialPatchIn(BaseModel):
description: Optional[str] = None
unit: Optional[str] = None
code: Optional[str] = None
active: Optional[bool] = None
class MaterialImportIn(BaseModel):
text: str = ""
dry_run: bool = False
class LocationIn(BaseModel):
level: str = "building"
parent_id: Optional[str] = None
name: str = ""
@app.post("/api/projects/{project_id}/locations")
def add_location(project_id: str, body: LocationIn,
user: models.User = Depends(auth.get_current_user),
db: Session = Depends(get_db)):
"""Add one value by hand. Same rules as the import — an import you cannot
correct afterwards is an import nobody trusts enough to run."""
if not db.get(models.Project, project_id):
raise HTTPException(status_code=404, detail="Project not found")
# Any project member, matching how the SOP baseline itself is authored — the
# Project Admin gate is on CHANGING a completed SOP, not on writing one.
require_project_access(db, user, project_id)
require_project_writable(db, user, project_id, "The location list cannot be changed")
name = (body.name or "").strip()
code = location_slug(name)
if not code:
raise HTTPException(status_code=400,
detail="That name has no letters or digits to make a code from")
level = (body.level or "").strip().lower()
if level not in LOCATION_LEVELS:
raise HTTPException(status_code=400,
detail="Level must be one of: " + ", ".join(LOCATION_LEVELS))
depth = LOCATION_LEVELS.index(level)
parent = None
if body.parent_id:
parent = db.get(models.LocationNode, body.parent_id)
if not parent or parent.project_id != project_id:
raise HTTPException(status_code=400, detail="Parent is not on this project")
if depth == 0 and parent is not None:
raise HTTPException(status_code=400, detail="A building has nothing above it")
if depth > 0 and parent is None:
raise HTTPException(status_code=400,
detail="A %s needs a %s above it" % (level, LOCATION_LEVELS[depth - 1]))
if parent is not None and LOCATION_LEVELS.index(parent.level) != depth - 1:
raise HTTPException(status_code=400,
detail="A %s cannot sit under a %s" % (level, parent.level))
path = (parent.path + "/" + code) if parent else code
clash = db.scalars(select(models.LocationNode).where(
(models.LocationNode.project_id == project_id) & (models.LocationNode.path == path)
)).first()
if clash:
if not clash.active:
clash.active = True
log_event(db, user, "location_reactivated", "project", project_id, project_id,
summary=path)
db.commit()
return clash.to_dict()
raise HTTPException(status_code=409, detail="“%s” is already on this project" % name)
top = db.scalar(select(func.max(models.LocationNode.sort)).where(
models.LocationNode.project_id == project_id)) or 0
node = models.LocationNode(
id=gen_id("loc"), project_id=project_id, parent_id=(parent.id if parent else None),
level=level, code=code, path=path, name=name, active=True, sort=top + 1,
created_by=user.username,
)
db.add(node)
log_event(db, user, "location_added", "project", project_id, project_id, summary=path)
db.commit()
return node.to_dict()
class LocationPatch(BaseModel):
name: Optional[str] = None
active: Optional[bool] = None
@app.patch("/api/projects/{project_id}/locations/{node_id}")
def update_location(project_id: str, node_id: str, body: LocationPatch,
user: models.User = Depends(auth.get_current_user),
db: Session = Depends(get_db)):
"""Rename or deactivate. There is no DELETE, and that is the design:
rename changes `name` only. `code` and `path` are untouched, so every
work package pointing at this value keeps pointing at it. That
is the whole reason the two are separate columns.
deactivate hides the value from new work packages. Cascades DOWN — a floor
nobody can pick makes its sectors unpickable too, and leaving
them offered would be offering a path to nowhere. Reactivating
a child reactivates its ancestors for the same reason.
"""
if not db.get(models.Project, project_id):
raise HTTPException(status_code=404, detail="Project not found")
# Any project member, matching how the SOP baseline itself is authored — the
# Project Admin gate is on CHANGING a completed SOP, not on writing one.
require_project_access(db, user, project_id)
require_project_writable(db, user, project_id, "The location list cannot be changed")
node = db.get(models.LocationNode, node_id)
if not node or node.project_id != project_id:
raise HTTPException(status_code=404, detail="Location not found on this project")
changed = {}
if body.name is not None:
name = body.name.strip()
if not name:
raise HTTPException(status_code=400, detail="A name cannot be empty")
if name != node.name:
changed["name"] = {"from": node.name, "to": name}
node.name = name
if body.active is not None and bool(body.active) != bool(node.active):
changed["active"] = {"from": bool(node.active), "to": bool(body.active)}
node.active = bool(body.active)
if not node.active:
for child in db.scalars(select(models.LocationNode).where(
(models.LocationNode.project_id == project_id)
& (models.LocationNode.path.like(node.path + "/%"))
)).all():
child.active = False
else:
# Walk up by path rather than by parent_id: one query, and it cannot
# loop on a malformed chain.
segs = node.path.split("/")
ancestors = ["/".join(segs[:n]) for n in range(1, len(segs))]
if ancestors:
for anc in db.scalars(select(models.LocationNode).where(
(models.LocationNode.project_id == project_id)
& (models.LocationNode.path.in_(ancestors))
)).all():
anc.active = True
if changed:
log_event(db, user, "location_updated", "project", project_id, project_id,
summary=node.path, detail=changed)
db.commit()
return node.to_dict()
@app.get("/api/projects/{project_id}/summary")
def project_summary(project_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
"""What the launcher needs to describe a project without asking the browser
what it remembers (B4).
The launcher used to read `wp_suite_sop_complete` out of localStorage, which is
a per-browser mirror: a colleague completing the SOP on their machine left your
card saying "Complete SOP first" with nothing to indicate the answer was stale."""
proj = db.get(models.Project, project_id)
if not proj:
raise HTTPException(status_code=404, detail="Project not found")
require_project_access(db, user, project_id)
sop = db.scalars(
select(models.Sop)
.where(models.Sop.project_id == project_id, models.Sop.complete.is_(True))
.order_by(models.Sop.updated_at.desc())
.limit(1)
).first()
wp_total = db.scalar(
select(func.count()).select_from(models.WorkPackage).where(
models.WorkPackage.project_id == project_id,
models.WorkPackage.archived_at.is_(None),
)
) or 0
return {
"project_id": project_id,
"project_name": proj.name,
"sop_complete": sop is not None,
"sop_id": sop.id if sop else None,
"sop_name": (sop.name if sop else "") or "",
"sop_updated_at": models._iso(sop.updated_at) if sop else None,
"wp_total": wp_total,
"generated_at": models.utcnow().isoformat(),
}
@app.get("/api/wps/{wp_id}")
def get_wp(wp_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
wp = db.get(models.WorkPackage, wp_id)
if not wp:
raise HTTPException(status_code=404, detail="Work Package not found")
require_project_access(db, user, wp.project_id)
return wp.to_dict()
@app.delete("/api/wps/{wp_id}")
def delete_wp(wp_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
wp = db.get(models.WorkPackage, wp_id)
if not wp:
raise HTTPException(status_code=404, detail="Work Package not found")
# Deleting a work package is irreversible — Project Admin only. A project_user
# who wants one out of the way can archive it instead (reversible).
require_project_admin(db, user, wp.project_id, "Deleting a work package")
require_project_writable(db, user, wp.project_id, "Deleting a work package")
log_event(db, user, "deleted", "wp", wp.id, project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id))
db.delete(wp)
db.commit()
return {"deleted": wp_id}
@app.post("/api/wps/{wp_id}/issue")
def issue_wp(wp_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
"""Release a Work Package to the field. Refuses while a release gate is unmet:
any open constraint, or a predecessor package that isn't Closed."""
wp = db.get(models.WorkPackage, wp_id)
if not wp:
raise HTTPException(status_code=404, detail="Work Package not found")
require_project_access(db, user, wp.project_id)
require_project_writable(db, user, wp.project_id, "Issuing a work package")
enforce_release_gates(db, wp.id, wp.data, "Issued", wp.status)
wp.status = "Issued"
wp.issued_at = models.utcnow()
log_event(db, user, "issued", "wp", wp.id, project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id), detail={"to": "Issued"})
db.commit()
db.refresh(wp)
return wp.to_dict()
@app.post("/api/wps/{wp_id}/status")
def set_wp_status(wp_id: str, body: StatusIn, background_tasks: BackgroundTasks, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
wp = db.get(models.WorkPackage, wp_id)
if not wp:
raise HTTPException(status_code=404, detail="Work Package not found")
require_project_access(db, user, wp.project_id)
require_project_writable(db, user, wp.project_id, "Changing a work package's status")
old_status = wp.status
_qa_notifs = []
# Same gates as /issue — this route must not be a way around them.
enforce_release_gates(db, wp.id, wp.data, body.status, old_status)
if old_status == QA_READY_STATUS and body.status == "In Progress":
comment = str(body.comment or "").strip()
if not comment:
raise HTTPException(status_code=409, detail={
"message": "Returning a package from Ready for QA requires a comment",
})
data = dict(wp.data or {})
rejs = [r for r in (data.get("qaRejections") or []) if isinstance(r, dict)]
rejs.append({"ts": models.utcnow().isoformat(), "comment": comment,
"by": user.full_name or user.username, "from": QA_READY_STATUS})
data["qaRejections"] = rejs
wp.data = data
wp.status = body.status
if body.status == "Issued" and wp.issued_at is None:
wp.issued_at = models.utcnow()
if old_status != body.status:
log_event(db, user, "status_changed", "wp", wp.id, project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id), detail={"from": old_status, "to": body.status})
if "Issue" in (old_status, body.status):
_hlds = [h for h in ((wp.data or {}).get("holds") or []) if isinstance(h, dict)]
if body.status == "Issue":
_h = next((h for h in reversed(_hlds) if not h.get("released")), None)
log_event(db, user, "hold_logged", "wp", wp.id, project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id),
detail={"from": old_status,
"constraint": str((_h or {}).get("constraint") or "")[:200],
"reason": str((_h or {}).get("details") or "")[:300]})
else:
_h = next((h for h in reversed(_hlds) if h.get("released")), None)
log_event(db, user, "hold_released", "wp", wp.id, project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id),
detail={"to": body.status,
"constraint": str((_h or {}).get("constraint") or "")[:200],
"reason": str((_h or {}).get("details") or "")[:300]})
if body.status == QA_READY_STATUS:
log_event(db, user, "qa_ready", "wp", wp.id, project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id), detail={"from": old_status})
_qa_notifs = notify_qa_transition(db, wp, user, rejected=False)
elif old_status == QA_READY_STATUS and body.status == "In Progress":
log_event(db, user, "qa_rejected", "wp", wp.id, project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id),
detail={"comment": qa_rejection_comment(wp.data)[:300]})
_qa_notifs = notify_qa_transition(db, wp, user, rejected=True)
db.commit()
db.refresh(wp)
for n in _qa_notifs:
background_tasks.add_task(notify.deliver, n.id)
return wp.to_dict()
@app.post("/api/wps/{wp_id}/archive")
def archive_wp(wp_id: str, body: ArchiveIn, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
"""Archive (or unarchive) a Work Package — hides it from the default lists and
the dashboard without deleting it. Kept for the record on long-running jobs."""
wp = db.get(models.WorkPackage, wp_id)
if not wp:
raise HTTPException(status_code=404, detail="Work Package not found")
require_project_access(db, user, wp.project_id)
# An archived project is frozen whole: its packages keep the archive state they
# had, and tidying inside it waits until the job is unarchived.
require_project_writable(db, user, wp.project_id, "Archiving a work package")
was_archived = wp.archived_at is not None
if body.archived and not was_archived:
wp.archived_at = models.utcnow()
log_event(db, user, "archived", "wp", wp.id, project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id))
elif not body.archived and was_archived:
wp.archived_at = None
log_event(db, user, "unarchived", "wp", wp.id, project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id))
db.commit()
db.refresh(wp)
return wp.to_dict()
# ── Audit trail (history) ──────────────────────────────────────────────────────
# ── Drawing uploads (CR-007 / D8) ──────────────────────────────────────────────
def project_storage_used(db: Session, project_id: str) -> int:
return int(db.scalar(
select(func.coalesce(func.sum(models.WpFile.size), 0))
.where(models.WpFile.project_id == project_id)) or 0)
def _wp_files_meta(db: Session, wp_id: str) -> list[dict]:
rows = db.scalars(select(models.WpFile).where(models.WpFile.wp_id == wp_id)
.order_by(models.WpFile.created_at)).all()
return [r.to_dict() for r in rows]
def _sync_wp_files(db: Session, wp: "models.WorkPackage") -> None:
"""Mirror the meta list into data["files"] - server-owned, so exports and the
offline cache read it straight off the package record."""
data = dict(wp.data or {})
data["files"] = _wp_files_meta(db, wp.id)
wp.data = data
@app.get("/api/projects/{project_id}/storage")
def project_storage(project_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
require_project_access(db, user, project_id)
used = project_storage_used(db, project_id)
return {"used": used, "ceiling": FILE_PROJECT_CEILING,
"warn_at": int(FILE_PROJECT_CEILING * 0.8)}
@app.get("/api/wps/{wp_id}/files")
def list_wp_files(wp_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
wp = db.get(models.WorkPackage, wp_id)
if not wp:
raise HTTPException(status_code=404, detail="Work Package not found")
require_project_access(db, user, wp.project_id)
used = project_storage_used(db, wp.project_id or "")
return {"files": _wp_files_meta(db, wp_id), "used": used,
"ceiling": FILE_PROJECT_CEILING}
@app.post("/api/wps/{wp_id}/files")
def upload_wp_file(wp_id: str, body: FileUploadIn, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
"""D8, enforced HERE, not only in the browser: 5MB a file, PDF or image, and
a per-project ceiling (2GB by default) that refuses by NAME when reached."""
wp = db.get(models.WorkPackage, wp_id)
if not wp:
raise HTTPException(status_code=404, detail="Work Package not found")
require_project_access(db, user, wp.project_id)
require_project_writable(db, user, wp.project_id, "Uploading a drawing")
mime = (body.mime or "").strip().lower()
if not FILE_ALLOWED_MIME_RE.match(mime):
raise HTTPException(status_code=400, detail={
"message": "Only PDF and image files are accepted", "mime": mime})
try:
raw = base64.b64decode(body.data_base64 or "", validate=True)
except Exception:
raise HTTPException(status_code=400, detail="File data is not valid base64")
if not raw:
raise HTTPException(status_code=400, detail="The file is empty")
if len(raw) > FILE_MAX_BYTES:
raise HTTPException(status_code=413, detail={
"message": "Files are limited to 5MB each", "size": len(raw),
"limit": FILE_MAX_BYTES})
used = project_storage_used(db, wp.project_id or "")
if used + len(raw) > FILE_PROJECT_CEILING:
raise HTTPException(status_code=413, detail={
"message": f"This project's drawing storage is full ({FILE_PROJECT_CEILING} bytes). "
"Remove an old drawing to make room.",
"used": used, "ceiling": FILE_PROJECT_CEILING})
row = models.WpFile(
id=gen_id("file"), wp_id=wp.id, project_id=wp.project_id or "",
name=(body.name or "drawing")[:300], mime=mime, size=len(raw),
description=(body.description or "")[:500], data=raw,
uploaded_by=user.full_name or user.username)
db.add(row)
db.flush()
_sync_wp_files(db, wp)
log_event(db, user, "file_uploaded", "wp", wp.id, project_id=wp.project_id,
summary=(wp.number or wp.subject or wp.id),
detail={"file": row.name, "size": row.size, "mime": row.mime})
db.commit()
used = project_storage_used(db, wp.project_id or "")
return {"file": row.to_dict(), "used": used, "ceiling": FILE_PROJECT_CEILING,
"warn": used >= int(FILE_PROJECT_CEILING * 0.8)}
@app.get("/api/files/{file_id}")
def get_wp_file(file_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
row = db.get(models.WpFile, file_id)
if not row:
raise HTTPException(status_code=404, detail="File not found")
require_project_access(db, user, row.project_id)
return Response(content=row.data, media_type=row.mime or "application/octet-stream",
headers={"Content-Disposition":
f'inline; filename="{(row.name or "file").replace(chr(34), "")}"',
"Cache-Control": "private, max-age=86400"})
@app.patch("/api/files/{file_id}")
def patch_wp_file(file_id: str, body: FileDescIn, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
row = db.get(models.WpFile, file_id)
if not row:
raise HTTPException(status_code=404, detail="File not found")
require_project_access(db, user, row.project_id)
require_project_writable(db, user, row.project_id, "Editing a drawing description")
row.description = (body.description or "")[:500]
wp = db.get(models.WorkPackage, row.wp_id)
if wp:
_sync_wp_files(db, wp)
db.commit()
return row.to_dict()
@app.delete("/api/files/{file_id}")
def delete_wp_file(file_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
row = db.get(models.WpFile, file_id)
if not row:
raise HTTPException(status_code=404, detail="File not found")
require_project_access(db, user, row.project_id)
require_project_writable(db, user, row.project_id, "Removing a drawing")
wp = db.get(models.WorkPackage, row.wp_id)
log_event(db, user, "file_deleted", "wp", row.wp_id, project_id=row.project_id,
summary=(wp.number if wp else row.wp_id),
detail={"file": row.name, "size": row.size})
db.delete(row)
db.flush()
if wp:
_sync_wp_files(db, wp)
db.commit()
used = project_storage_used(db, row.project_id)
return {"deleted": file_id, "used": used, "ceiling": FILE_PROJECT_CEILING}
# ── Project material list (D6 / T8.6) - the CR-005 pattern, for materials ─────
def parse_material_rows(text: str):
"""description[,unit[,code]] per row - comma, semicolon or tab separated, a
paste straight out of a spreadsheet. A header row is ignored. Rejections come
back with the SOURCE line number: an import that says '42 rows' over a file
with 50 has lost eight and told nobody."""
rows, rejected = [], []
header_words = ("description", "desc", "item", "material")
for i, raw in enumerate((text or "").split(chr(10)), start=1):
# rstrip only: a LEADING separator means the first column is empty, and
# the first column is the description - eating it would accept ",FT" as
# a material named FT (found by the probe on the first run).
line = raw.strip().rstrip(",;")
if not line:
continue
parts = [p.strip() for p in re.split(r"[,;\t]", line)]
if i == 1 and parts and parts[0].lower() in header_words:
continue
parts = [p for p in parts]
if not parts or not parts[0]:
rejected.append({"line": i, "text": raw.strip()[:120],
"reason": "no description in the first column"})
continue
if len(parts) > 3:
rejected.append({"line": i, "text": raw.strip()[:120],
"reason": "more than three columns - description, unit, code is the whole shape"})
continue
rows.append((i, parts))
return rows, rejected
def material_key(desc: str, code: str) -> str:
return (code or "").strip().lower() or location_slug(desc).lower()
@app.get("/api/projects/{project_id}/materials")
def list_materials(project_id: str, include_inactive: bool = Query(False),
user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
require_project_access(db, user, project_id)
stmt = select(models.MaterialItem).where(models.MaterialItem.project_id == project_id)
if not include_inactive:
stmt = stmt.where(models.MaterialItem.active.is_(True))
rows = db.scalars(stmt.order_by(models.MaterialItem.sort, models.MaterialItem.description)).all()
return {"items": [r.to_dict() for r in rows]}
@app.post("/api/projects/{project_id}/materials/import")
def import_materials(project_id: str, body: MaterialImportIn,
user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
if not db.get(models.Project, project_id):
raise HTTPException(status_code=404, detail="Project not found")
require_project_access(db, user, project_id)
require_project_writable(db, user, project_id, "The material list cannot be changed")
rows, rejected = parse_material_rows(body.text)
existing = {material_key(r.description, r.code): r for r in db.scalars(
select(models.MaterialItem).where(models.MaterialItem.project_id == project_id)).all()}
created, duplicates, reactivated = [], [], []
seen_in_file = {}
next_sort = max((r.sort for r in existing.values()), default=0)
for line_no, parts in rows:
desc = parts[0][:300]
unit = (parts[1] if len(parts) > 1 else "")[:20].upper()
code = (parts[2] if len(parts) > 2 else "")[:80]
key = material_key(desc, code)
if key in seen_in_file:
duplicates.append({"line": line_no, "text": desc,
"reason": "already on line %d of this import" % seen_in_file[key]})
continue
seen_in_file[key] = line_no
if key in existing:
row = existing[key]
if not row.active:
if not body.dry_run:
row.active = True
reactivated.append({"text": desc})
else:
duplicates.append({"line": line_no, "text": desc,
"reason": "already on this project"})
continue
next_sort += 1
created.append({"description": desc, "unit": unit, "code": code})
if not body.dry_run:
item = models.MaterialItem(id=gen_id("mat"), project_id=project_id,
code=code, description=desc, unit=unit,
active=True, sort=next_sort)
db.add(item)
existing[key] = item
if not body.dry_run:
log_event(db, user, "materials_imported", "project", project_id,
project_id=project_id, summary="material list",
detail={"created": len(created), "rejected": len(rejected)})
db.commit()
return {"read": len(rows) + len(rejected), "created": created,
"rejected": rejected, "duplicates": duplicates,
"reactivated": reactivated, "dry_run": body.dry_run}
@app.post("/api/projects/{project_id}/materials")
def add_material(project_id: str, body: MaterialIn,
user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
if not db.get(models.Project, project_id):
raise HTTPException(status_code=404, detail="Project not found")
require_project_access(db, user, project_id)
require_project_writable(db, user, project_id, "The material list cannot be changed")
desc = (body.description or "").strip()
if not desc:
raise HTTPException(status_code=400, detail="A description is required.")
key = material_key(desc, body.code)
clash = [r for r in db.scalars(select(models.MaterialItem)
.where(models.MaterialItem.project_id == project_id)).all()
if material_key(r.description, r.code) == key]
if clash:
raise HTTPException(status_code=409, detail="That material is already on this project.")
next_sort = (db.scalar(select(func.coalesce(func.max(models.MaterialItem.sort), 0))
.where(models.MaterialItem.project_id == project_id)) or 0) + 1
item = models.MaterialItem(id=gen_id("mat"), project_id=project_id,
code=(body.code or "").strip()[:80],
description=desc[:300],
unit=(body.unit or "").strip()[:20].upper(),
active=True, sort=next_sort)
db.add(item)
db.commit()
return item.to_dict()
@app.patch("/api/projects/{project_id}/materials/{item_id}")
def patch_material(project_id: str, item_id: str, body: MaterialPatchIn,
user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
row = db.get(models.MaterialItem, item_id)
if not row or row.project_id != project_id:
raise HTTPException(status_code=404, detail="Material not found")
require_project_access(db, user, project_id)
require_project_writable(db, user, project_id, "The material list cannot be changed")
if body.description is not None:
row.description = body.description.strip()[:300]
if body.unit is not None:
row.unit = body.unit.strip()[:20].upper()
if body.code is not None:
row.code = body.code.strip()[:80]
if body.active is not None:
# Deactivate, never delete - a request already referencing the line must
# keep rendering it (the CR-005 rule, applied to materials).
row.active = bool(body.active)
db.commit()
return row.to_dict()
@app.get("/api/audit")
def list_audit(
entity_type: Optional[str] = Query(None),
entity_id: Optional[str] = Query(None),
project_id: Optional[str] = Query(None),
action: Optional[str] = Query(None),
limit: int = Query(100, ge=1, le=500),
offset: int = Query(0, ge=0),
user: models.User = Depends(auth.get_current_user),
db: Session = Depends(get_db),
):
"""History / audit trail. Pass entity_type+entity_id for one item's history,
or project_id for a project activity feed. Scoped to the caller's project
access; admins additionally see project-less events (user management)."""
stmt = select(models.AuditLog)
if entity_type:
stmt = stmt.where(models.AuditLog.entity_type == entity_type)
if entity_id:
stmt = stmt.where(models.AuditLog.entity_id == entity_id)
if action:
stmt = stmt.where(models.AuditLog.action == action)
if project_id:
require_project_access(db, user, project_id)
stmt = stmt.where(models.AuditLog.project_id == project_id)
# Non-admins only ever see events tied to a project they can access.
ids = accessible_project_ids(db, user)
if ids is not None:
stmt = stmt.where(models.AuditLog.project_id.in_(ids))
rows = db.scalars(stmt.order_by(models.AuditLog.at.desc()).limit(limit).offset(offset)).all()
return [e.to_dict() for e in rows]
# ── Global search ──────────────────────────────────────────────────────────────
def _like_term(q: str) -> str:
"""Escape LIKE wildcards so a user searching for '100%' or 'a_b' gets what they
typed rather than a pattern. Paired with escape='\\' on the comparison."""
return "%" + q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_").lower() + "%"
@app.get("/api/search")
def global_search(
q: str = Query("", min_length=0, max_length=200),
limit: int = Query(8, ge=1, le=25),
user: models.User = Depends(auth.get_current_user),
db: Session = Depends(get_db),
):
"""Type-ahead across projects, work packages and SOPs, scoped to the projects
the caller may access. Matches work-package number, subject, type and status,
project name/number/client/site, and SOP name/number."""
term = (q or "").strip()
if len(term) < 2:
return {"query": term, "projects": [], "wps": [], "sops": []}
pat = _like_term(term)
esc = "\\"
# An archived project is meant to be gone from view, and search is the one place
# it would otherwise come back — as the project itself, or as one of its packages
# or SOPs. So the archived ids are read once here and filtered out of all three
# result sets. Rows with no project at all (orphan/legacy, admin-only) survive
# explicitly: SQL's NOT IN drops NULLs, and they aren't on an archived job.
archived_pids = set(db.scalars(
select(models.Project.id).where(models.Project.archived_at.is_not(None))
).all())
proj_stmt = select(models.Project).where(
func.lower(models.Project.name).like(pat, escape=esc)
| func.lower(models.Project.number).like(pat, escape=esc)
| func.lower(models.Project.client).like(pat, escape=esc)
| func.lower(models.Project.site).like(pat, escape=esc)
)
if archived_pids:
proj_stmt = proj_stmt.where(models.Project.id.not_in(archived_pids))
proj_stmt = scope_to_access(proj_stmt, models.Project.id, db, user)
projects = db.scalars(proj_stmt.order_by(models.Project.updated_at.desc()).limit(limit)).all()
wp_stmt = select(models.WorkPackage).where(
(models.WorkPackage.archived_at.is_(None))
& (
func.lower(models.WorkPackage.number).like(pat, escape=esc)
| func.lower(models.WorkPackage.subject).like(pat, escape=esc)
| func.lower(models.WorkPackage.type).like(pat, escape=esc)
| func.lower(models.WorkPackage.status).like(pat, escape=esc)
)
)
if archived_pids:
wp_stmt = wp_stmt.where(
models.WorkPackage.project_id.is_(None)
| models.WorkPackage.project_id.not_in(archived_pids)
)
wp_stmt = scope_to_access(wp_stmt, models.WorkPackage.project_id, db, user)
wps = db.scalars(wp_stmt.order_by(models.WorkPackage.updated_at.desc()).limit(limit)).all()
sop_stmt = select(models.Sop).where(
func.lower(models.Sop.name).like(pat, escape=esc)
| func.lower(models.Sop.number).like(pat, escape=esc)
)
if archived_pids:
sop_stmt = sop_stmt.where(
models.Sop.project_id.is_(None) | models.Sop.project_id.not_in(archived_pids)
)
sop_stmt = scope_to_access(sop_stmt, models.Sop.project_id, db, user)
sops = db.scalars(sop_stmt.order_by(models.Sop.updated_at.desc()).limit(limit)).all()
# Project names for the WP/SOP rows, so a result reads unambiguously when the
# same WP number exists on two jobs.
pids = {w.project_id for w in wps} | {s.project_id for s in sops}
pids.discard(None)
names = {}
if pids:
for p in db.scalars(select(models.Project).where(models.Project.id.in_(pids))).all():
names[p.id] = p.name or p.number or p.id
return {
"query": term,
"projects": [{"id": p.id, "name": p.name, "number": p.number, "client": p.client} for p in projects],
"wps": [{
"id": w.id, "number": w.number, "subject": w.subject, "type": w.type,
"status": w.status, "project_id": w.project_id,
"project_name": names.get(w.project_id, ""),
} for w in wps],
"sops": [{
"id": s.id, "name": s.name, "number": s.number, "complete": s.complete,
"project_id": s.project_id, "project_name": names.get(s.project_id, ""),
} for s in sops],
}
# ── Settings (admin) ────────────────────────────────────────────────────────────
@app.get("/api/settings")
def get_app_settings(_admin: models.User = Depends(auth.require_admin), db: Session = Depends(get_db)):
return notify.public_settings(db)
@app.get("/api/app-flags")
def get_app_flags(_user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
"""Feature flags every signed-in page reads (e.g. whether the BIM/VDC tooling
is switched on). No secrets — safe for any authenticated user."""
return notify.app_flags(db)
@app.put("/api/settings")
def put_app_settings(body: SettingsIn, admin: models.User = Depends(auth.require_admin), db: Session = Depends(get_db)):
patch = {k: v for k, v in body.model_dump().items() if v is not None}
# Localization defaults are validated the same way a user's own preference is,
# so a typo can't leave every page formatting dates against a bogus zone.
loc = (patch.get("default_locale") or "").strip()
if loc and not _LOCALE_RE.match(loc):
raise HTTPException(status_code=400, detail="Default locale must be a language tag like 'en-US'.")
tz = (patch.get("default_timezone") or "").strip()
if tz and not valid_timezone(tz):
raise HTTPException(status_code=400, detail="Unknown default time zone.")
if "default_locale" in patch:
patch["default_locale"] = loc
if "default_timezone" in patch:
patch["default_timezone"] = tz
saved = notify.save_settings(db, patch)
log_event(db, admin, "settings_updated", "settings", "notifications",
summary="notifications", detail={"email_enabled": bool(saved.get("email_enabled"))})
db.commit()
return notify.public_settings(db)
@app.post("/api/settings/test-email")
def send_test_email(body: TestEmailIn, admin: models.User = Depends(auth.require_admin), db: Session = Depends(get_db)):
s = notify.get_settings(db)
if not notify.smtp_ready(s):
raise HTTPException(status_code=400, detail="Set the SMTP host and From address first.")
to = (body.to or admin.email or "").strip()
if not to:
raise HTTPException(status_code=400, detail="No recipient — add an email to your account or pass 'to'.")
try:
notify.send_email(s, to, "Work Package Suite — test email",
"This is a test from the Work Package Suite. If you got this, SMTP is working.")
except Exception as e: # noqa: BLE001
raise HTTPException(status_code=400, detail=f"Send failed: {e}")
return {"ok": True, "to": to}
# ── Notifications + project members ─────────────────────────────────────────────
@app.get("/api/notifications")
def list_notifications(all: bool = Query(False), limit: int = Query(100, ge=1, le=500),
user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
stmt = select(models.Notification)
if not (all and auth.is_admin(user)):
stmt = stmt.where(models.Notification.user_id == user.id)
rows = db.scalars(stmt.order_by(models.Notification.created_at.desc()).limit(limit)).all()
return [n.to_dict() for n in rows]
@app.get("/api/projects/{project_id}/members")
def project_members(project_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
"""Users who can be assigned WPs on this project — its members plus admins."""
require_project_access(db, user, project_id)
member_ids = set(db.scalars(select(models.ProjectMember.user_id).where(models.ProjectMember.project_id == project_id)).all())
members = db.scalars(select(models.User).where(models.User.id.in_(member_ids))).all() if member_ids else []
admins = db.scalars(select(models.User).where(models.User.role == auth.ROLE_ADMIN)).all()
out, seen = [], set()
for u in list(members) + list(admins):
if u.id in seen or not u.is_active:
continue
seen.add(u.id)
out.append({"id": u.id, "username": u.username, "full_name": u.full_name,
"email": u.email, "project_role": u.project_role or "",
"role": effective_role(db, u, project_id)})
out.sort(key=lambda x: (x["full_name"] or x["username"] or "").lower())
return out
# ── Comments / feedback ──────────────────────────────────────────────────────
def _save_comment(body: CommentIn, db: Session, user: "models.User") -> dict:
# A comment tied to a WP/SOP requires access to that resource's project, so
# a user can't write into another project's review thread. A review thread on an
# archived project is frozen with the rest of it; general app feedback isn't tied
# to a project at all and keeps working regardless.
#
# Both ids are checked independently — NOT if/elif. The row stores whichever ids
# the payload carried, so a body naming a WP you may touch AND a SOP you may not
# would, under an elif, be authorised on the WP alone and still land in the other
# project's SOP thread.
check_id(body.wp_id)
check_id(body.sop_id)
if body.wp_id:
wp = db.get(models.WorkPackage, body.wp_id)
if not wp:
raise HTTPException(status_code=404, detail="Work Package not found")
require_project_access(db, user, wp.project_id)
require_project_writable(db, user, wp.project_id, "Commenting on a work package")
if body.sop_id:
sop = db.get(models.Sop, body.sop_id)
if not sop:
raise HTTPException(status_code=404, detail="SOP not found")
require_project_access(db, user, sop.project_id)
require_project_writable(db, user, sop.project_id, "Commenting on a SOP")
extra = body.model_extra or {}
c = models.Comment(
id=gen_id("c"),
source=body.source or body.type or "",
sop_id=body.sop_id,
wp_id=body.wp_id,
step=body.step,
# Attribution comes from the authenticated session, NEVER the client
# payload — otherwise comments could be forged as another user.
author=(user.full_name or user.username),
text=body.text or "",
page=body.page or "",
extra=extra,
)
db.add(c)
db.commit()
db.refresh(c)
return c.to_dict()
@app.post("/api/comments")
def create_comment(body: CommentIn, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
return _save_comment(body, db, user)
# Alias so the existing client (which posts to /api/feedback) keeps working.
@app.post("/api/feedback")
def create_feedback(body: CommentIn, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
return _save_comment(body, db, user)
@app.get("/api/comments")
def list_comments(
source: Optional[str] = Query(None),
sop_id: Optional[str] = Query(None),
wp_id: Optional[str] = Query(None),
step: Optional[int] = Query(None),
user: models.User = Depends(auth.get_current_user),
db: Session = Depends(get_db),
):
stmt = select(models.Comment)
if source:
stmt = stmt.where(models.Comment.source == source)
if sop_id:
stmt = stmt.where(models.Comment.sop_id == sop_id)
if wp_id:
stmt = stmt.where(models.Comment.wp_id == wp_id)
if step is not None:
stmt = stmt.where(models.Comment.step == step)
# Non-admins see general app feedback plus comments on WPs/SOPs in their own
# projects only — never another project's review threads.
ids = accessible_project_ids(db, user)
if ids is not None:
acc_wp = select(models.WorkPackage.id).where(models.WorkPackage.project_id.in_(ids))
acc_sop = select(models.Sop.id).where(models.Sop.project_id.in_(ids))
# The "general feedback" branch is ONLY for comments not tied to any
# WP/SOP — otherwise a project-scoped comment tagged source=home_feedback
# by the client would leak across projects. Project-tied comments are
# visible strictly by project membership.
stmt = stmt.where(
((models.Comment.source == "home_feedback")
& models.Comment.wp_id.is_(None) & models.Comment.sop_id.is_(None))
| (models.Comment.wp_id.in_(acc_wp))
| (models.Comment.sop_id.in_(acc_sop))
)
rows = db.scalars(stmt.order_by(models.Comment.created_at.desc())).all()
return [c.to_dict() for c in rows]
# ── Micron asset catalog (read-only lookup) ────────────────────────────────────
# Backs the asset picker in the work package creator. This is a *lookup*, not a
# resource this app owns: there is no POST, and nothing here ever writes to the
# Micron database. It is deliberately not project-scoped by the app's own access
# rules — the catalog is reference data, and any signed-in user who can build a
# work package needs to be able to name the assets it covers. Authentication is
# still required (the auth_gate middleware covers every /api/ path).
@app.get("/api/assets")
def list_assets(_user: models.User = Depends(auth.get_current_user)):
"""The whole catalog, fetched once when the creator loads. Searching happens
in the browser — there is no per-keystroke endpoint by design."""
if not assets_db.configured():
# Not an error — the suite is designed to run without Micron wired up.
# The picker reads this and switches to manual entry.
return {"configured": False, "assets": [], "detail": assets_db.status()["detail"]}
try:
return {"configured": True, "assets": assets_db.load()}
except assets_db.AssetSourceError as exc:
# 503, not 500: the suite is healthy, its upstream lookup is not. The
# picker degrades to manual entry rather than blocking the package.
raise HTTPException(status_code=503, detail=str(exc))
# ── Local dev convenience: serve the static site from this app ──────────────────
# In production NGINX serves html/ and only proxies /api/ here, so this app never
# receives "/" requests, and the api Docker image doesn't even include html/ — so
# this mount stays inactive there. Locally (plain uvicorn, no NGINX) it lets you
# open the whole suite at http://localhost:8000/ with the API on the SAME origin,
# so the session cookie just works (no CORS, no Secure-cookie headache).
#
# Mounted LAST so the /api/* routes above always match first.
_html_dir = os.path.join(os.path.dirname(__file__), "..", "html")
if os.path.isdir(_html_dir):
class _NoCacheCode(StaticFiles):
"""Serve code assets with Cache-Control: no-cache.
Production runs behind NGINX (which now sets this itself), but the dev server
is what people actually click around in — and with no header at all the
browser applies HEURISTIC freshness per file (~10% of the file's age), so the
least recently changed file gets the longest lifetime and HTML/CSS/JS drift
apart between reloads. ETag/Last-Modified still make revalidation a cheap 304.
"""
async def get_response(self, path, scope):
res = await super().get_response(path, scope)
if path.endswith((".html", ".css", ".js", ".webmanifest")) or path in ("", "/", "."):
res.headers["Cache-Control"] = "no-cache"
return res
app.mount("/", _NoCacheCode(directory=_html_dir, html=True), name="site")
Reference in New Issue View Git Blame Copy Permalink