"""The navigation tree: categories, what is filed under them, and the order.
Two levels and nothing else. A `Category` is a named group; a `Subcategory` is a
leaf under one, or under none - and "under none" is a null pointer rather than a
row, so there is no "Ungrouped" category to seed in every organisation and no way
for two representations of the same idea to disagree.
The tree is the organisation's, not a person's: everyone in a tenant reads the
same sidebar. Arranging it therefore needs `Scope.ADMIN` - the scope the skeleton
already reserves for administrative acts - while reading it is ordinary READ.
That is one rule, reusing what is here, rather than a second permission model
alongside the first.
What each person keeps to themselves is which twisties they folded, and that
lives in app/core/nav_state.py.
Rename all of it. `Category` and `Subcategory` are placeholders for whatever your
product files things under - clients and projects, teams and boards, departments
and reports. The shape is what ports; the nouns are yours.
"""
import logging
import re
from sqlalchemy import func
from db.db_declarations import Category, Subcategory
from .context import Conflict, Invalid, NotFound, Scope
logger = logging.getLogger(__name__)
NAME_MAX = 60
# "Not passed", so a partial move leaves the other half alone. None cannot say
# that here: it is a value, and it means ungrouped.
_UNSET = object()
_SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,58}[a-z0-9]$|^[a-z0-9]$")
def can_arrange(principal) -> bool:
"""Whether this caller may add, rename, reorder or delete.
Answered for the shell as well as enforced below, so the sidebar can leave
the controls off for somebody the server would refuse - offering an
affordance and then rejecting the request it produces is a worse answer than
not offering it.
It is not a substitute for the check. Every write below requires ADMIN on its
own, because a control the shell declined to draw is not a gate.
"""
return Scope.ADMIN in principal.scopes
# --- categories -------------------------------------------------------------
def list_categories(session, principal) -> list[Category]:
"""Every category in this organisation, in the order the sidebar shows them.
Ranks first, then id: a category created before ranks were assigned - or two
that ended up sharing one - still lands somewhere stable and repeatable
rather than wherever the database felt like returning it.
"""
principal.require(Scope.READ)
return (session.query(Category)
.filter(Category.org_id == principal.org_id)
.order_by(Category.rank.is_(None), Category.rank, Category.id)
.all())
def get_category(session, principal, category_id) -> Category:
"""One category, scoped to the caller's organisation.
NotFound rather than Forbidden for an id belonging to another tenant: "that
exists but is not yours" confirms a competitor is a customer.
"""
principal.require(Scope.READ)
try:
category_id = int(category_id)
except (TypeError, ValueError):
raise NotFound("No such category.")
category = (session.query(Category)
.filter(Category.org_id == principal.org_id,
Category.id == category_id)
.one_or_none())
if category is None:
raise NotFound("No such category.")
return category
def create_category(session, principal, name: str) -> Category:
"""Add a category at the bottom of the list - where the control that made it
is."""
principal.require(Scope.ADMIN)
category = Category(org_id=principal.org_id,
name=_clean_name(session, principal, Category, name),
rank=_next_rank(session, principal.org_id, Category))
session.add(category)
session.commit()
return category
def update_category(session, principal, category_id: int, *,
name=None, rank=None) -> Category:
"""Rename a category, or move it among its siblings.
Only what is passed changes, so a reorder and a rename are one endpoint
without either silently clearing the other's field.
"""
principal.require(Scope.ADMIN)
category = get_category(session, principal, category_id)
if name is not None:
category.name = _clean_name(session, principal, Category, name,
exclude_id=category.id)
if rank is not None:
category.rank = _as_rank(rank)
session.commit()
return category
def delete_category(session, principal, category_id: int) -> dict:
"""Remove a category. What was filed under it becomes ungrouped.
Nothing is deleted beyond the heading - this is a decision about a row in a
list, not about the things in it. The count comes back so the shell can say
where they went.
"""
principal.require(Scope.ADMIN)
category = get_category(session, principal, category_id)
# Cleared explicitly rather than left to ON DELETE SET NULL: SQLite enforces
# no foreign key unless asked, so a cascade-only clear would leave the test
# database pointing at a category that no longer exists while Postgres did
# the right thing. That is the divergence you only find in production.
#
# nav_rank goes with it: a rank is only meaningful among the siblings it was
# measured against, and these rows have just changed siblings.
freed = (session.query(Subcategory)
.filter(Subcategory.org_id == principal.org_id,
Subcategory.category_id == category.id)
.update({Subcategory.category_id: None, Subcategory.nav_rank: None},
synchronize_session=False))
name = category.name
session.delete(category)
session.commit()
logger.info("Deleted category %r; %d item(s) became ungrouped.", name, freed)
return {"id": category_id, "name": name, "ungrouped": freed}
# --- subcategories ----------------------------------------------------------
def list_subcategories(session, principal) -> list[Subcategory]:
"""Every subcategory in this organisation, ordered as the sidebar reads
them: rank, then name for anything unranked."""
principal.require(Scope.READ)
return (session.query(Subcategory)
.filter(Subcategory.org_id == principal.org_id)
.order_by(Subcategory.nav_rank.is_(None), Subcategory.nav_rank,
Subcategory.name)
.all())
def get_subcategory(session, principal, slug: str) -> Subcategory:
principal.require(Scope.READ)
row = (session.query(Subcategory)
.filter(Subcategory.org_id == principal.org_id,
Subcategory.slug == (slug or "").strip().lower())
.one_or_none())
if row is None:
raise NotFound("No such item.")
return row
def create_subcategory(session, principal, name: str, *,
slug: str | None = None, category_id=None) -> Subcategory:
"""Add a leaf, optionally filed under a category."""
principal.require(Scope.ADMIN)
row = Subcategory(
org_id=principal.org_id,
name=_clean_name(session, principal, Subcategory, name),
slug=_clean_slug(session, principal, slug or name),
category_id=(get_category(session, principal, category_id).id
if category_id is not None else None),
nav_rank=_next_rank(session, principal.org_id, Subcategory),
)
session.add(row)
session.commit()
return row
def update_subcategory(session, principal, slug: str, *, name=None) -> Subcategory:
"""Rename a leaf.
The slug deliberately does not change with it: renaming a thing should not
break every link that already points at it.
"""
principal.require(Scope.ADMIN)
row = get_subcategory(session, principal, slug)
if name is not None:
row.name = _clean_name(session, principal, Subcategory, name,
exclude_id=row.id)
session.commit()
return row
def delete_subcategory(session, principal, slug: str) -> dict:
principal.require(Scope.ADMIN)
row = get_subcategory(session, principal, slug)
name = row.name
session.delete(row)
session.commit()
return {"slug": slug, "name": name}
def place_subcategory(session, principal, slug: str, *,
category_id=_UNSET, nav_rank=_UNSET) -> Subcategory:
"""File a leaf under a category (or under none), at a rank among its
neighbours.
One write covering both halves of a move, because they are one gesture: which
group it is in, and where it sits inside that group. Either half may be
omitted and is then left alone; `None` is a value here, not an omission, and
means ungrouped.
"""
principal.require(Scope.ADMIN)
row = get_subcategory(session, principal, slug)
if category_id is not _UNSET:
# Resolved through get_category, so an id from another tenant reads as
# "no such category" rather than as a refusal that confirms it exists.
row.category_id = (get_category(session, principal, category_id).id
if category_id is not None else None)
if nav_rank is not _UNSET:
row.nav_rank = _as_rank(nav_rank) if nav_rank is not None else None
session.commit()
return row
# --- what the sidebar renders -----------------------------------------------
def tree(session, principal) -> dict:
"""The whole navigation tree, grouped, in two queries.
Built here rather than in the template so the grouping rule - and in
particular what "ungrouped" means - is written once, where it can be tested
without rendering HTML. Two queries and not one per category, because a
sidebar with twenty groups should not cost twenty round trips.
"""
principal.require(Scope.READ)
categories = list_categories(session, principal)
rows = list_subcategories(session, principal)
grouped: dict[int | None, list] = {c.id: [] for c in categories}
grouped[None] = []
for row in rows:
# A row pointing at a category that no longer exists reads as ungrouped
# rather than vanishing from a list meant to show everything.
key = row.category_id if row.category_id in grouped else None
grouped[key].append(row)
return {
"categories": [{"id": c.id, "name": c.name, "items": grouped[c.id]}
for c in categories],
# Not a category: it has no row on the server, cannot be renamed or
# deleted, and is not in the reorder. It is simply where the rest are.
"ungrouped": grouped[None],
"can_arrange": can_arrange(principal),
}
# --- helpers ----------------------------------------------------------------
def _clean_name(session, principal, model, name: str, *, exclude_id=None) -> str:
name = (name or "").strip()
if not name:
raise Invalid("It needs a name.")
if len(name) > NAME_MAX:
raise Invalid(f"A name is at most {NAME_MAX} characters.")
# Checked here as well as by the unique constraint, so the answer is a
# sentence rather than the constraint's generic 409. The constraint is still
# what enforces it under a race.
clash = (session.query(model.id)
.filter(model.org_id == principal.org_id, model.name == name))
if exclude_id is not None:
clash = clash.filter(model.id != exclude_id)
if clash.first() is not None:
raise Conflict(f"There is already something called {name!r}.")
return name
def _clean_slug(session, principal, raw: str) -> str:
slug = re.sub(r"[^a-z0-9]+", "-", (raw or "").strip().lower()).strip("-")
if not _SLUG_RE.match(slug):
raise Invalid("That does not make a usable URL. Use letters and digits.")
if (session.query(Subcategory.id)
.filter(Subcategory.org_id == principal.org_id,
Subcategory.slug == slug).first() is not None):
raise Conflict(f"There is already an item at {slug!r}.")
return slug
def _as_rank(value) -> float:
try:
return float(value)
except (TypeError, ValueError):
raise Invalid("A rank must be a number.")
def _next_rank(session, org_id: int, model) -> float:
"""One past the last, so a new row lands at the bottom of the list - where
the control that made it is."""
column = model.rank if model is Category else model.nav_rank
last = (session.query(func.max(column))
.filter(model.org_id == org_id).scalar())
return (last or 0) + 1