peter
8206384c91
The owner-filter check missed an aliased() model, an import alias, .in_() and filter_by; four live inline filters on aliased VideoState go through owned_by now. Cross-unit writes are seen under any import name, through sa.update and via __table__. Core modules may commit only where CORE_COMMITS names the reason; trim_read and audit.gc flush now and their callers commit. The pin target, the debug-session header and a link's target fetch through owned_row. The docs claim what the fence holds, and say what is beyond it.
92 lines
3.4 KiB
Python
92 lines
3.4 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. A contract test
|
|
(`test_unit_fences.py`) holds every module, core included, to the filter's form; a rule this module
|
|
cannot say (own or shared, built-in or own) is written out where it applies, with its reason.
|
|
|
|
**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())
|