bvandeusen
0f80b790c7
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 9s
CI & Build / integration (push) Successful in 19s
CI & Build / TypeScript typecheck (push) Failing after 20s
CI & Build / Python tests (push) Successful in 44s
CI & Build / Build & push image (push) Skipped
Last piece of the architecture in #2296. The operator: "the prose doesn't have to live as one offs, there's a central system for managing it." Two fields, both free-form: design_systems.guidance the narrative a token table cannot hold — aesthetic, voice and tone, what is deliberately out of scope. design_tokens.rationale WHY a token is this value, which is a different question from `purpose` (what it is FOR). "Success equals Moss, aligned by design" is a rationale; "page bg, deepest surface" is a purpose. Rules carry the first routinely and a token row had nowhere to put it. Free-form rather than a column per category, deliberately. A schema with `voice`, `aesthetic` and `scope` columns would bake one rulebook's table of contents into every install (rule #115), leaving the next install three empty columns and nowhere for what it actually cares about. Both nullable: a design system with no prose at all is complete, not a draft. `rationale` cascades like `purpose` — deepest non-empty wins — so an app overriding a colour keeps the family's reasoning rather than blanking it. Same argument as `supersedes`: the override was about the value, not the meaning. In the generated sheet the inline comment prefers `purpose` and falls back to `rationale`, so a token carrying only the why still says something instead of rendering bare.
502 lines
19 KiB
Python
502 lines
19 KiB
Python
"""Design system + token persistence, and the guard on the parent chain.
|
|
|
|
Access goes through `services/access.py` (rule #78), where reaching a system via
|
|
a project you can see grants READ but never write — see
|
|
`get_design_system_permission` for why that asymmetry exists.
|
|
|
|
The cascade itself lives in `services/design_cascade.py` as pure functions. This
|
|
module is the part that needs a database: loading the shape of the hierarchy,
|
|
refusing writes that would break it, and storing tokens.
|
|
|
|
Nothing here seeds or implies a default system. An install with no design
|
|
systems is an ordinary install, and every caller must handle an empty list as
|
|
the normal case rather than a missing prerequisite.
|
|
"""
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import select
|
|
|
|
from scribe.models import async_session
|
|
from scribe.models.design_system import DesignSystem, DesignToken
|
|
from scribe.models.project import Project
|
|
from scribe.services import access
|
|
from scribe.services.design_stylesheet import (
|
|
check_code_against_tokens,
|
|
duplicate_values,
|
|
render_stylesheet,
|
|
)
|
|
from scribe.services.design_cascade import (
|
|
ResolvedToken,
|
|
ancestry,
|
|
resolve_tokens,
|
|
would_cycle,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DesignSystemCycle(ValueError):
|
|
"""A parent assignment that would close an inheritance loop.
|
|
|
|
Raised rather than returned as None because the two outcomes need different
|
|
answers: None already means "not found, or not yours", and a caller that
|
|
conflated them would show "no such design system" for what is really "that
|
|
parent is one of its own descendants". Routes map this to a 400.
|
|
"""
|
|
|
|
|
|
async def _parent_map(session, owner_user_id: int) -> dict[int, int | None]:
|
|
"""`{id: parent_id}` for one owner's live systems — the hierarchy's shape.
|
|
|
|
Scoped to the OWNER of the systems, not the caller, and the distinction is
|
|
load-bearing on the read path: a caller reading through a shared project
|
|
does not own any link in the chain, and a caller-scoped map would hand them
|
|
an empty forest and a cascade truncated to one system. They would get a page
|
|
that renders with plausible wrong values and no error anywhere.
|
|
|
|
Owner-scoping is also the constraint on parenting: a chain may only be built
|
|
from systems its owner controls, or someone else's delete or re-parent would
|
|
silently restyle your app.
|
|
"""
|
|
rows = (
|
|
await session.execute(
|
|
select(DesignSystem.id, DesignSystem.parent_id).where(
|
|
DesignSystem.owner_user_id == owner_user_id,
|
|
DesignSystem.deleted_at.is_(None),
|
|
)
|
|
)
|
|
).all()
|
|
return dict(rows)
|
|
|
|
|
|
# --- design systems ---------------------------------------------------------
|
|
|
|
async def create_design_system(
|
|
user_id: int,
|
|
title: str,
|
|
description: str | None = None,
|
|
guidance: str | None = None,
|
|
parent_id: int | None = None,
|
|
) -> DesignSystem | None:
|
|
"""Create a system, with or without a parent.
|
|
|
|
Returns None when `parent_id` names a system the caller may not write —
|
|
which, per the ACL, means one they do not own.
|
|
"""
|
|
if parent_id is not None and not await access.can_write_design_system(
|
|
user_id, parent_id
|
|
):
|
|
return None
|
|
async with async_session() as session:
|
|
system = DesignSystem(
|
|
owner_user_id=user_id,
|
|
title=title.strip(),
|
|
description=description,
|
|
guidance=guidance,
|
|
parent_id=parent_id,
|
|
)
|
|
session.add(system)
|
|
await session.commit()
|
|
await session.refresh(system)
|
|
return system
|
|
|
|
|
|
async def get_design_system(user_id: int, design_system_id: int) -> DesignSystem | None:
|
|
async with async_session() as session:
|
|
system = await session.get(DesignSystem, design_system_id)
|
|
if system is None or system.deleted_at is not None:
|
|
return None
|
|
if not await access.can_read_design_system(user_id, design_system_id):
|
|
return None
|
|
return system
|
|
|
|
|
|
async def list_design_systems(user_id: int) -> list[DesignSystem]:
|
|
"""The caller's own systems, ordered by title. Empty is normal."""
|
|
async with async_session() as session:
|
|
rows = await session.execute(
|
|
select(DesignSystem)
|
|
.where(
|
|
DesignSystem.owner_user_id == user_id,
|
|
DesignSystem.deleted_at.is_(None),
|
|
)
|
|
.order_by(DesignSystem.title)
|
|
)
|
|
return list(rows.scalars().all())
|
|
|
|
|
|
async def update_design_system(
|
|
user_id: int, design_system_id: int, **fields: object
|
|
) -> DesignSystem | None:
|
|
"""Partial update. Raises DesignSystemCycle if `parent_id` would loop.
|
|
|
|
`parent_id` is handled apart from the other fields because None is a
|
|
meaningful value for it — "make this a root" — where for every other field
|
|
None means "leave alone". Callers signal it by passing the key at all.
|
|
"""
|
|
if not await access.can_write_design_system(user_id, design_system_id):
|
|
return None
|
|
async with async_session() as session:
|
|
system = await session.get(DesignSystem, design_system_id)
|
|
if system is None or system.deleted_at is not None:
|
|
return None
|
|
|
|
if "parent_id" in fields:
|
|
parent_id = fields.pop("parent_id")
|
|
if parent_id is not None:
|
|
parent_id = int(parent_id)
|
|
if not await access.can_write_design_system(user_id, parent_id):
|
|
return None
|
|
if would_cycle(
|
|
design_system_id,
|
|
parent_id,
|
|
await _parent_map(session, system.owner_user_id),
|
|
):
|
|
raise DesignSystemCycle(
|
|
f"Design system {design_system_id} cannot inherit from "
|
|
f"{parent_id}: that system already inherits from it."
|
|
)
|
|
system.parent_id = parent_id
|
|
|
|
for key, value in fields.items():
|
|
if key in ("title", "description", "guidance") and value is not None:
|
|
setattr(system, key, value)
|
|
system.updated_at = datetime.now(timezone.utc)
|
|
await session.commit()
|
|
await session.refresh(system)
|
|
return system
|
|
|
|
|
|
async def delete_design_system(user_id: int, design_system_id: int) -> bool:
|
|
"""Soft-delete a system. Children survive as roots (the FK is SET NULL)."""
|
|
if not await access.can_write_design_system(user_id, design_system_id):
|
|
return False
|
|
async with async_session() as session:
|
|
system = await session.get(DesignSystem, design_system_id)
|
|
if system is None or system.deleted_at is not None:
|
|
return False
|
|
system.deleted_at = datetime.now(timezone.utc)
|
|
await session.commit()
|
|
return True
|
|
|
|
|
|
# --- tokens -----------------------------------------------------------------
|
|
|
|
async def create_token(
|
|
user_id: int,
|
|
design_system_id: int,
|
|
name: str,
|
|
value_by_mode: dict | None = None,
|
|
group_name: str | None = None,
|
|
purpose: str | None = None,
|
|
rationale: str | None = None,
|
|
supersedes: list | None = None,
|
|
order_index: int = 0,
|
|
) -> DesignToken | None:
|
|
if not await access.can_write_design_system(user_id, design_system_id):
|
|
return None
|
|
async with async_session() as session:
|
|
token = DesignToken(
|
|
design_system_id=design_system_id,
|
|
name=name.strip(),
|
|
# `or {}` and not the argument as given: the column is NOT NULL so
|
|
# that absence has exactly one spelling. Passing None here would
|
|
# otherwise store JSON null and reintroduce the second empty state.
|
|
value_by_mode=value_by_mode or {},
|
|
group_name=group_name,
|
|
purpose=purpose,
|
|
rationale=rationale,
|
|
# `or []` for the same reason as value_by_mode above: the column is
|
|
# NOT NULL so absence has one spelling, and None would store JSON
|
|
# null instead of an empty array.
|
|
supersedes=supersedes or [],
|
|
order_index=order_index,
|
|
)
|
|
session.add(token)
|
|
await session.commit()
|
|
await session.refresh(token)
|
|
return token
|
|
|
|
|
|
async def list_tokens(user_id: int, design_system_id: int) -> list[DesignToken]:
|
|
"""One system's OWN tokens — its override set, not its effective set.
|
|
|
|
Resolving the chain is step 2's job; this deliberately answers the narrower
|
|
question ("what does this system change?") that the model exists to make
|
|
free.
|
|
"""
|
|
if not await access.can_read_design_system(user_id, design_system_id):
|
|
return []
|
|
async with async_session() as session:
|
|
rows = await session.execute(
|
|
select(DesignToken)
|
|
.where(
|
|
DesignToken.design_system_id == design_system_id,
|
|
DesignToken.deleted_at.is_(None),
|
|
)
|
|
.order_by(DesignToken.order_index.asc(), DesignToken.name.asc())
|
|
)
|
|
return list(rows.scalars().all())
|
|
|
|
|
|
async def resolve_design_system(
|
|
user_id: int, design_system_id: int
|
|
) -> list[ResolvedToken] | None:
|
|
"""A system's EFFECTIVE token set — everything it inherits, with its own on top.
|
|
|
|
None when the caller may not read the system; an empty list when the chain
|
|
genuinely holds no tokens, which is an ordinary state for a system that has
|
|
just been created.
|
|
|
|
Two queries regardless of how deep the chain runs: one for the hierarchy's
|
|
shape, one for every token in it. The flattening itself is
|
|
`design_cascade.resolve_tokens` — pure, so the cascade rule is tested
|
|
without a database and this function is only the loading.
|
|
"""
|
|
if not await access.can_read_design_system(user_id, design_system_id):
|
|
return None
|
|
async with async_session() as session:
|
|
system = await session.get(DesignSystem, design_system_id)
|
|
if system is None or system.deleted_at is not None:
|
|
return None
|
|
|
|
parents = await _parent_map(session, system.owner_user_id)
|
|
chain = ancestry(design_system_id, parents)
|
|
|
|
rows = (
|
|
await session.execute(
|
|
select(DesignToken)
|
|
.where(
|
|
DesignToken.design_system_id.in_(chain),
|
|
DesignToken.deleted_at.is_(None),
|
|
)
|
|
.order_by(DesignToken.order_index.asc(), DesignToken.name.asc())
|
|
)
|
|
).scalars().all()
|
|
|
|
tokens_by_system: dict[int, list[DesignToken]] = {}
|
|
for token in rows:
|
|
tokens_by_system.setdefault(token.design_system_id, []).append(token)
|
|
return resolve_tokens(design_system_id, parents, tokens_by_system)
|
|
|
|
|
|
async def update_token(
|
|
user_id: int, token_id: int, **fields: object
|
|
) -> DesignToken | None:
|
|
allowed = {
|
|
"name", "value_by_mode", "group_name", "purpose", "rationale",
|
|
"supersedes", "order_index",
|
|
}
|
|
async with async_session() as session:
|
|
token = await session.get(DesignToken, token_id)
|
|
if token is None or token.deleted_at is not None:
|
|
return None
|
|
if not await access.can_write_design_system(user_id, token.design_system_id):
|
|
return None
|
|
for key, value in fields.items():
|
|
if key in allowed and value is not None:
|
|
setattr(token, key, value)
|
|
token.updated_at = datetime.now(timezone.utc)
|
|
await session.commit()
|
|
await session.refresh(token)
|
|
return token
|
|
|
|
|
|
async def delete_token(user_id: int, token_id: int) -> bool:
|
|
async with async_session() as session:
|
|
token = await session.get(DesignToken, token_id)
|
|
if token is None or token.deleted_at is not None:
|
|
return False
|
|
if not await access.can_write_design_system(user_id, token.design_system_id):
|
|
return False
|
|
token.deleted_at = datetime.now(timezone.utc)
|
|
await session.commit()
|
|
return True
|
|
|
|
|
|
# --- the project pointer ----------------------------------------------------
|
|
|
|
async def set_project_design_system(
|
|
user_id: int, project_id: int, design_system_id: int | None
|
|
) -> bool:
|
|
"""Point a project at a design system, or at nothing (None clears it).
|
|
|
|
Needs write on the project and READ on the system: pointing at a system is
|
|
consuming it, not changing it, so a system shared with you through another
|
|
project is a legitimate choice here.
|
|
"""
|
|
if not await access.can_write_project(user_id, project_id):
|
|
return False
|
|
if design_system_id is not None and not await access.can_read_design_system(
|
|
user_id, design_system_id
|
|
):
|
|
return False
|
|
async with async_session() as session:
|
|
project = await session.get(Project, project_id)
|
|
if project is None or project.deleted_at is not None:
|
|
return False
|
|
project.design_system_id = design_system_id
|
|
project.updated_at = datetime.now(timezone.utc)
|
|
await session.commit()
|
|
return True
|
|
|
|
|
|
# --- import from a rulebook -------------------------------------------------
|
|
|
|
async def import_from_rulebook(
|
|
user_id: int,
|
|
design_system_id: int,
|
|
rulebook_id: int,
|
|
apply: bool = False,
|
|
) -> dict | None:
|
|
"""Propose (and optionally create) tokens for a system from a rulebook.
|
|
|
|
Returns None when the caller may not write the system or read the rulebook.
|
|
Otherwise a report with three lists, and the split between them is the whole
|
|
point of running it with `apply=False` first:
|
|
|
|
proposed — everything the rulebook describes, each entry carrying the
|
|
rule and sentence it came from
|
|
created — what was actually written (empty unless `apply`)
|
|
skipped — proposals whose name the system already defines
|
|
|
|
**Existing tokens are never overwritten.** An import is a proposal built by
|
|
reading prose; a value already in the record was put there deliberately, and
|
|
a re-run must not undo an operator's correction. That also makes the whole
|
|
operation safe to repeat — it fills gaps and reports the rest.
|
|
|
|
Tokens with no value are still created when `apply` is set. The rulebook
|
|
names them, so their absence from the system is itself a finding, and a
|
|
named token with a blank value says "this exists and needs deciding" where
|
|
silence says nothing at all.
|
|
"""
|
|
if not await access.can_write_design_system(user_id, design_system_id):
|
|
return None
|
|
|
|
from scribe.services import rulebooks as rulebooks_svc
|
|
from scribe.services.design_rulebook_import import propose_tokens
|
|
|
|
rules = await rulebooks_svc.list_rules(user_id, rulebook_id=rulebook_id)
|
|
if not rules:
|
|
return {"rulebook_id": rulebook_id, "proposed": [], "created": [], "skipped": []}
|
|
|
|
proposals = propose_tokens(rules)
|
|
existing = {t.name for t in await list_tokens(user_id, design_system_id)}
|
|
|
|
created: list[dict] = []
|
|
skipped: list[str] = []
|
|
for index, proposal in enumerate(proposals):
|
|
if proposal.name in existing:
|
|
skipped.append(proposal.name)
|
|
continue
|
|
if not apply:
|
|
continue
|
|
token = await create_token(
|
|
user_id,
|
|
design_system_id=design_system_id,
|
|
name=proposal.name,
|
|
value_by_mode=proposal.value_by_mode,
|
|
group_name=proposal.group_name,
|
|
purpose=proposal.purpose,
|
|
supersedes=proposal.supersedes,
|
|
order_index=index,
|
|
)
|
|
if token is not None:
|
|
created.append(token.to_dict())
|
|
|
|
return {
|
|
"rulebook_id": rulebook_id,
|
|
"proposed": [p.as_dict() for p in proposals],
|
|
"created": created,
|
|
"skipped": skipped,
|
|
}
|
|
|
|
|
|
# --- the master sheet -------------------------------------------------------
|
|
|
|
async def stylesheet_for_system(
|
|
user_id: int, design_system_id: int, root_selector: str = ":root"
|
|
) -> dict | None:
|
|
"""The master CSS sheet a design system generates, plus its reuse report.
|
|
|
|
Purpose tokens only — the sheet styles no elements. See
|
|
`services/design_stylesheet.py` for why that split is the design rather than
|
|
a shortcut.
|
|
|
|
Returns None if the caller may not read the system. The `duplicates` half is
|
|
advisory: two tokens sharing a value are either a deliberate alias or one
|
|
idea recorded twice, and only the operator knows which.
|
|
"""
|
|
resolved = await resolve_design_system(user_id, design_system_id)
|
|
if resolved is None:
|
|
return None
|
|
system = await get_design_system(user_id, design_system_id)
|
|
|
|
return {
|
|
"design_system_id": design_system_id,
|
|
"css": render_stylesheet(
|
|
resolved,
|
|
root_selector=root_selector,
|
|
title=system.title if system else "",
|
|
design_system_id=design_system_id,
|
|
),
|
|
"token_count": len(resolved),
|
|
"valueless": [t.name for t in resolved if not t.value_by_mode],
|
|
"duplicates": duplicate_values(resolved),
|
|
}
|
|
|
|
|
|
async def check_snippets_against_system(
|
|
user_id: int, design_system_id: int, project_id: int = 0
|
|
) -> dict | None:
|
|
"""Which recorded snippets disagree with this design system's sheet.
|
|
|
|
The relation the operator named — "the snippets use the tags from the sheet"
|
|
— turned into a check. For each snippet: `var(--x)` references with no such
|
|
token, literals the sheet says to stop writing, and custom properties the
|
|
snippet mints for itself instead of using shared ones.
|
|
|
|
Snippets with nothing to report are omitted entirely. A list of everything
|
|
that is fine is a list nobody reads twice.
|
|
"""
|
|
resolved = await resolve_design_system(user_id, design_system_id)
|
|
if resolved is None:
|
|
return None
|
|
|
|
from scribe.services import snippets as snippets_svc
|
|
|
|
# `list_snippets` returns (rows, total) and caps limit at 100; `project_id`
|
|
# must be None — not 0 — to reach across every project, since 0 would filter
|
|
# to a project with that id.
|
|
rows, _total = await snippets_svc.list_snippets(
|
|
user_id=user_id, project_id=project_id or None, limit=100,
|
|
)
|
|
|
|
findings: list[dict] = []
|
|
for row in rows:
|
|
snippet_id = row.get("id")
|
|
if snippet_id is None:
|
|
continue
|
|
# The list rows carry a preview, not the code. The check has to read the
|
|
# whole body or it would report on a truncation.
|
|
note = await snippets_svc.get_snippet(user_id=user_id, snippet_id=int(snippet_id))
|
|
if note is None:
|
|
continue
|
|
report = check_code_against_tokens(note.body or "", resolved)
|
|
if not (
|
|
report["unknown"] or report["superseded_literals"] or report["local_definitions"]
|
|
):
|
|
continue
|
|
findings.append({
|
|
"snippet_id": int(snippet_id),
|
|
"title": note.title or "",
|
|
**report,
|
|
})
|
|
|
|
return {
|
|
"design_system_id": design_system_id,
|
|
"checked": len(rows),
|
|
"findings": findings,
|
|
}
|