peter
5ce292cd28
The per-file pin list and the unit-by-unit switch go: every route file is held to zero database work, every unit's package to its edges, and core code says whose a row is through userscope too. CLAUDE.md, the backend README and the gate register describe the units as they now are.
91 lines
3.3 KiB
Python
91 lines
3.3 KiB
Python
"""Whose a row is, and who an account is to the others — one place for both.
|
|
|
|
**Ownership.** A user-owned table says whose a row is in one column (`user_id`, or `to_user_id` for
|
|
something handed to them). Every read and write names that column through `owned_by` and fetches a
|
|
row by id through `owned_row`, never with an inline `Model.user_id == …`: the rule "a row owned by
|
|
someone else is answered like a missing one" then lives here, and a contract test
|
|
(`test_unit_fences.py`) holds every module to it, core included.
|
|
|
|
**Recipients.** "An active, non-suspended account" — the download share-by-email and the recipient
|
|
picker filter on it. Kept here so the two forms (SQL clauses for a query, a Python check for a loaded
|
|
row) can't drift apart.
|
|
"""
|
|
|
|
from typing import Any, TypeVar
|
|
|
|
from sqlalchemy import ColumnElement, func, select
|
|
from sqlalchemy.orm import InstrumentedAttribute, Session
|
|
|
|
from app.collation import collated
|
|
from app.db import Base
|
|
from app.models import User
|
|
|
|
M = TypeVar("M", bound=Base)
|
|
|
|
|
|
def _owner_id(owner: User | int) -> int:
|
|
return owner if isinstance(owner, int) else owner.id
|
|
|
|
|
|
def owned_by(column: InstrumentedAttribute, owner: User | int) -> ColumnElement[bool]:
|
|
"""`column` — a table's owner column — names `owner`."""
|
|
return column == _owner_id(owner)
|
|
|
|
|
|
def owned_row(
|
|
db: Session,
|
|
model: type[M],
|
|
id_: Any,
|
|
owner: User | int,
|
|
*,
|
|
owner_attr: str = "user_id",
|
|
lock: bool = False,
|
|
) -> M | None:
|
|
"""The row of `model` with this primary key when `owner` owns it, else None — a row of someone
|
|
else's is the same answer as a missing one, so a caller can never tell the two apart. `lock`
|
|
takes it FOR UPDATE (past the identity map)."""
|
|
row = db.get(model, id_, with_for_update=True if lock else None)
|
|
if row is None or getattr(row, owner_attr) != _owner_id(owner):
|
|
return None
|
|
return row
|
|
|
|
|
|
def messageable_clauses() -> list:
|
|
"""SQL clauses for "an active, non-suspended account" — splat into a recipient query."""
|
|
return [
|
|
User.is_active.is_(True),
|
|
User.is_suspended.is_(False),
|
|
]
|
|
|
|
|
|
def is_messageable_user(u: User | None) -> bool:
|
|
"""Python mirror of `messageable_clauses()` for an already-loaded row (a share recipient)."""
|
|
return bool(u and u.is_active and not u.is_suspended)
|
|
|
|
|
|
def user_by_email(db: Session, email: str) -> User | None:
|
|
"""The account with this address, matched without regard to case."""
|
|
return db.execute(
|
|
select(User).where(func.lower(User.email) == email.strip().lower())
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def messageable_others(db: Session, user: User, lang: str) -> list[User]:
|
|
"""Every account `user` may hand something to: active, not suspended, not themselves; in the
|
|
reader's alphabet by address."""
|
|
return list(
|
|
db.execute(
|
|
select(User)
|
|
.where(User.id != user.id, *messageable_clauses())
|
|
.order_by(collated(User.email, lang))
|
|
).scalars()
|
|
)
|
|
|
|
|
|
def emails_of(db: Session, ids: set[int] | None = None) -> dict[int, str]:
|
|
"""Each account's address by id, in address order: these ids, or every account."""
|
|
stmt = select(User.id, User.email).order_by(User.email)
|
|
if ids is not None:
|
|
stmt = stmt.where(User.id.in_(ids))
|
|
return dict(db.execute(stmt).tuples().all())
|