267 lines
11 KiB
Python
267 lines
11 KiB
Python
 
|
"""Departments and their stations, as the Admin edits them.
|
||
|
|
|
||
|
|
Code and slug never change (the group names derive from them; a trigger refuses it too). Stations are never deleted,
|
||
|
|
because station limits and packets point at them: `active: false` hides one. A new department is a shop department
|
||
|
|
with the three shop groups IT must create in the PacketStation OU. The support departments map to fixed roles, so a
|
||
|
|
new one would have no group and no role; it is refused.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import re
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from psycopg.rows import dict_row
|
||
|
|
|
||
|
|
from .. import audit
|
||
|
|
from ..api.errors import ApiError
|
||
|
|
from ..auth import groups as groupmap
|
||
|
|
|
||
|
|
RX_CODE = re.compile(r"^[0-9A-Z]{2,4}$")
|
||
|
|
RX_SLUG = re.compile(r"^[A-Za-z][A-Za-z0-9]{1,30}$")
|
||
|
|
NAME_MAX = 80
|
||
|
|
STATION_NAME_MAX = 60
|
||
|
|
STATIONS_MAX = 60
|
||
|
|
|
||
|
|
|
||
|
|
def clean_name(value: Any, what: str, limit: int) -> str:
|
||
|
|
text = " ".join(str(value or "").split())
|
||
|
|
if not text:
|
||
|
|
raise ApiError(422, "invalid_request", f"Give the {what} a name.")
|
||
|
|
if len(text) > limit:
|
||
|
|
raise ApiError(422, "invalid_request", f"The {what} name is longer than {limit} characters.")
|
||
|
|
return text
|
||
|
|
|
||
|
|
|
||
|
|
def group_names_for(code: str, slug: str, kind: str, supervisor_group: bool) -> list[tuple[str, str]]:
|
||
|
|
"""(group name, role) for one department, as the group map builds them."""
|
||
|
|
if kind == "shop":
|
||
|
|
return [(f"PS-{code}-{slug}-{lv.capitalize()}", f"shop:{code}:{lv}") for lv in groupmap.SHOP_LEVELS]
|
||
|
|
kinds = groupmap.OFFICE_KINDS.get(code)
|
||
|
|
if kinds is None:
|
||
|
|
return []
|
||
|
|
member, sup = kinds
|
||
|
|
out = [(f"PS-{slug}", member)]
|
||
|
|
if supervisor_group and sup:
|
||
|
|
out.append((f"PS-{slug}-Supervisor", sup))
|
||
|
|
return out
|
||
|
|
|
||
|
|
|
||
|
|
def _stations(conn: Any, code: str | None = None) -> dict[str, list[dict[str, Any]]]:
|
||
|
|
sql = "SELECT dept_code, id, name, sort, active FROM stations"
|
||
|
|
params: tuple[Any, ...] = ()
|
||
|
|
if code is not None:
|
||
|
|
sql += " WHERE dept_code = %s"
|
||
|
|
params = (code,)
|
||
|
|
out: dict[str, list[dict[str, Any]]] = {}
|
||
|
|
for dept, sid, name, sort, active in conn.execute(sql + " ORDER BY dept_code, sort, name, id", params):
|
||
|
|
out.setdefault(dept, []).append({"id": int(sid), "name": name, "sort": int(sort), "active": bool(active)})
|
||
|
|
return out
|
||
|
|
|
||
|
|
|
||
|
|
def _out(r: dict[str, Any], stations: list[dict[str, Any]], in_ou: set[str] | None) -> dict[str, Any]:
|
||
|
|
groups = [
|
||
|
|
{"name": name, "role": role, "in_ou": None if in_ou is None else name.lower() in in_ou}
|
||
|
|
for name, role in group_names_for(r["code"], r["slug"], r["kind"], bool(r["supervisor_group"]))
|
||
|
|
]
|
||
|
|
return {
|
||
|
|
"code": r["code"],
|
||
|
|
"slug": r["slug"],
|
||
|
|
"name": r["name"],
|
||
|
|
"kind": r["kind"],
|
||
|
|
"sort": int(r["sort"]),
|
||
|
|
"active": bool(r["active"]),
|
||
|
|
"weld": bool(r["weld"]),
|
||
|
|
"supervisor_group": bool(r["supervisor_group"]),
|
||
|
|
"created_at": r["created_at"],
|
||
|
|
"groups": groups,
|
||
|
|
"stations": stations,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
SELECT = (
|
||
|
|
"SELECT code, slug, name, kind::text AS kind, sort, active, weld, supervisor_group, created_at FROM departments"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def list_all(conn: Any, in_ou: list[str] | None) -> list[dict[str, Any]]:
|
||
|
|
"""Every department, active or not, in sort order. `in_ou` is the group names in the OU (None: unknown)."""
|
||
|
|
lower = {n.lower() for n in in_ou} if in_ou is not None else None
|
||
|
|
with conn.cursor(row_factory=dict_row) as cur:
|
||
|
|
rows = cur.execute(SELECT + " ORDER BY sort, code").fetchall()
|
||
|
|
st = _stations(conn)
|
||
|
|
return [_out(r, st.get(r["code"], []), lower) for r in rows]
|
||
|
|
|
||
|
|
|
||
|
|
def get(conn: Any, code: str, in_ou: list[str] | None = None) -> dict[str, Any] | None:
|
||
|
|
lower = {n.lower() for n in in_ou} if in_ou is not None else None
|
||
|
|
with conn.cursor(row_factory=dict_row) as cur:
|
||
|
|
r = cur.execute(SELECT + " WHERE code = %s", (code,)).fetchone()
|
||
|
|
if r is None:
|
||
|
|
return None
|
||
|
|
return _out(r, _stations(conn, code).get(code, []), lower)
|
||
|
|
|
||
|
|
|
||
|
|
def _snapshot(d: dict[str, Any]) -> dict[str, Any]:
|
||
|
|
return {k: d[k] for k in ("name", "active", "stations")}
|
||
|
|
|
||
|
|
|
||
|
|
def update(
|
||
|
|
conn: Any,
|
||
|
|
who: Any,
|
||
|
|
code: str,
|
||
|
|
*,
|
||
|
|
name: str | None,
|
||
|
|
active: bool | None,
|
||
|
|
stations: list[dict[str, Any]] | None,
|
||
|
|
) -> dict[str, Any]:
|
||
|
|
"""Change the name, active and stations of one department. Audited `admin.department_updated`."""
|
||
|
|
with conn.transaction():
|
||
|
|
locked = conn.execute("SELECT code FROM departments WHERE code = %s FOR UPDATE", (code,)).fetchone()
|
||
|
|
if locked is None:
|
||
|
|
raise ApiError(404, "not_found", "No such department.")
|
||
|
|
before = get(conn, code)
|
||
|
|
assert before is not None
|
||
|
|
if name is not None:
|
||
|
|
conn.execute(
|
||
|
|
"UPDATE departments SET name = %s WHERE code = %s", (clean_name(name, "department", NAME_MAX), code)
|
||
|
|
)
|
||
|
|
if active is not None:
|
||
|
|
conn.execute("UPDATE departments SET active = %s WHERE code = %s", (bool(active), code))
|
||
|
|
if stations is not None:
|
||
|
|
_apply_stations(conn, code, before["stations"], stations)
|
||
|
|
after = get(conn, code)
|
||
|
|
assert after is not None
|
||
|
|
if _snapshot(before) != _snapshot(after):
|
||
|
|
audit.append(
|
||
|
|
conn,
|
||
|
|
action="admin.department_updated",
|
||
|
|
user_id=who.user_id,
|
||
|
|
session_id=who.session_id,
|
||
|
|
role_snapshot=who.role_snapshot,
|
||
|
|
subject_type="department",
|
||
|
|
subject_id=code,
|
||
|
|
detail={"before": _snapshot(before), "after": _snapshot(after)},
|
||
|
|
ip=who.ip,
|
||
|
|
)
|
||
|
|
return after
|
||
|
|
|
||
|
|
|
||
|
|
def _apply_stations(conn: Any, code: str, current: list[dict[str, Any]], wanted: list[dict[str, Any]]) -> None:
|
||
|
|
by_id = {s["id"]: s for s in current}
|
||
|
|
final: dict[int, dict[str, Any]] = {s["id"]: dict(s) for s in current}
|
||
|
|
new: list[dict[str, Any]] = []
|
||
|
|
seen_ids: set[int] = set()
|
||
|
|
for w in wanted:
|
||
|
|
sid = w.get("id")
|
||
|
|
entry = {
|
||
|
|
"name": clean_name(w.get("name"), "station", STATION_NAME_MAX),
|
||
|
|
"sort": int(w.get("sort") or 0),
|
||
|
|
"active": bool(w.get("active", True)),
|
||
|
|
}
|
||
|
|
if sid is None:
|
||
|
|
new.append(entry)
|
||
|
|
continue
|
||
|
|
if sid not in by_id:
|
||
|
|
raise ApiError(422, "invalid_request", f"Station {sid} is not in department {code}.")
|
||
|
|
if sid in seen_ids:
|
||
|
|
raise ApiError(422, "invalid_request", f"Station {sid} is listed twice.")
|
||
|
|
seen_ids.add(sid)
|
||
|
|
final[sid].update(entry)
|
||
|
|
names = [s["name"].lower() for s in final.values()] + [s["name"].lower() for s in new]
|
||
|
|
dupes = sorted({n for n in names if names.count(n) > 1})
|
||
|
|
if dupes:
|
||
|
|
raise ApiError(422, "duplicate_station", f"Two stations in {code} have the same name: {', '.join(dupes)}.")
|
||
|
|
if len(final) + len(new) > STATIONS_MAX:
|
||
|
|
raise ApiError(422, "invalid_request", f"A department has at most {STATIONS_MAX} stations.")
|
||
|
|
renamed = [sid for sid, s in final.items() if s["name"] != by_id[sid]["name"]]
|
||
|
|
# Names are unique per department: park the renamed ones first so two stations can swap names.
|
||
|
|
for sid in renamed:
|
||
|
|
conn.execute("UPDATE stations SET name = %s WHERE id = %s", (f"~{sid}~renaming", sid))
|
||
|
|
for sid, s in final.items():
|
||
|
|
if s != by_id[sid]:
|
||
|
|
conn.execute(
|
||
|
|
"UPDATE stations SET name = %s, sort = %s, active = %s WHERE id = %s",
|
||
|
|
(s["name"], s["sort"], s["active"], sid),
|
||
|
|
)
|
||
|
|
for s in new:
|
||
|
|
conn.execute(
|
||
|
|
"INSERT INTO stations (dept_code, name, sort, active) VALUES (%s, %s, %s, %s)",
|
||
|
|
(code, s["name"], s["sort"], s["active"]),
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def add(
|
||
|
|
conn: Any,
|
||
|
|
who: Any,
|
||
|
|
*,
|
||
|
|
code: str,
|
||
|
|
slug: str,
|
||
|
|
name: str,
|
||
|
|
kind: str,
|
||
|
|
weld: bool,
|
||
|
|
supervisor_group: bool,
|
||
|
|
stations: list[str],
|
||
|
|
) -> tuple[dict[str, Any], list[str]]:
|
||
|
|
"""Add a shop department. Returns (department, group names IT must create). Audited `admin.department_added`."""
|
||
|
|
code, slug = code.strip(), slug.strip()
|
||
|
|
if not RX_CODE.match(code):
|
||
|
|
raise ApiError(422, "invalid_request", "The code is 2 to 4 capital letters or digits, like 15.")
|
||
|
|
if not RX_SLUG.match(slug):
|
||
|
|
raise ApiError(422, "invalid_request", "The short name is a letter, then letters or digits, no spaces.")
|
||
|
|
if kind != "shop":
|
||
|
|
raise ApiError(
|
||
|
|
422,
|
||
|
|
"support_fixed",
|
||
|
|
"Only a shop department can be added. The support departments map to fixed roles.",
|
||
|
|
)
|
||
|
|
title = clean_name(name, "department", NAME_MAX)
|
||
|
|
names: list[str] = []
|
||
|
|
for s in stations:
|
||
|
|
n = clean_name(s, "station", STATION_NAME_MAX)
|
||
|
|
if n.lower() in (x.lower() for x in names):
|
||
|
|
raise ApiError(422, "duplicate_station", f"Two stations have the same name: {n}.")
|
||
|
|
names.append(n)
|
||
|
|
if len(names) > STATIONS_MAX:
|
||
|
|
raise ApiError(422, "invalid_request", f"A department has at most {STATIONS_MAX} stations.")
|
||
|
|
with conn.transaction():
|
||
|
|
conn.execute("LOCK TABLE departments IN SHARE ROW EXCLUSIVE MODE")
|
||
|
|
if conn.execute("SELECT 1 FROM departments WHERE upper(code) = upper(%s)", (code,)).fetchone():
|
||
|
|
raise ApiError(409, "code_taken", f"Department code {code} is taken.")
|
||
|
|
if conn.execute("SELECT 1 FROM departments WHERE lower(slug) = lower(%s)", (slug,)).fetchone():
|
||
|
|
raise ApiError(409, "slug_taken", f"The short name {slug} is taken.")
|
||
|
|
sort = int(conn.execute("SELECT coalesce(max(sort), 0) + 10 FROM departments").fetchone()[0])
|
||
|
|
conn.execute(
|
||
|
|
"INSERT INTO departments (code, slug, name, kind, sort, active, supervisor_group, weld, created_at,"
|
||
|
|
" created_by) VALUES (%s, %s, %s, %s::dept_kind, %s, true, %s, %s, now(), %s)",
|
||
|
|
(code, slug, title, kind, sort, bool(supervisor_group), bool(weld), who.user_id),
|
||
|
|
)
|
||
|
|
for i, n in enumerate(names, start=1):
|
||
|
|
conn.execute("INSERT INTO stations (dept_code, name, sort) VALUES (%s, %s, %s)", (code, n, i))
|
||
|
|
dept = get(conn, code)
|
||
|
|
assert dept is not None
|
||
|
|
groups = [g["name"] for g in dept["groups"]]
|
||
|
|
audit.append(
|
||
|
|
conn,
|
||
|
|
action="admin.department_added",
|
||
|
|
user_id=who.user_id,
|
||
|
|
session_id=who.session_id,
|
||
|
|
role_snapshot=who.role_snapshot,
|
||
|
|
subject_type="department",
|
||
|
|
subject_id=code,
|
||
|
|
detail={
|
||
|
|
"after": {
|
||
|
|
"code": code,
|
||
|
|
"slug": slug,
|
||
|
|
"name": title,
|
||
|
|
"kind": kind,
|
||
|
|
"weld": bool(weld),
|
||
|
|
"supervisor_group": bool(supervisor_group),
|
||
|
|
"stations": names,
|
||
|
|
},
|
||
|
|
"groups_to_create": groups,
|
||
|
|
},
|
||
|
|
ip=who.ip,
|
||
|
|
)
|
||
|
|
return dept, groups
|