"""nw.genres — production *genres*: the reusable specialization of a project.
A **Genre** is a pure-data descriptor of a *kind* of audiovisual production
(music video, narrative video, commentary weave, music visualizer, ...). It is
the first-class formalization of what nw informally called an "app": a bundle
declared *over the substrate that already exists*, carrying no engine of its
own.
A genre bundles, by reference:
- ``body_schema_uris`` — the lacing body schemas (``annot://schema/<kind>/vN``)
its artifacts are validated against;
- ``transform_names`` — the :mod:`nw.transforms` entries forming its pipeline DAG;
- ``strategy_names`` — the optional :mod:`nw.renderers` strategies it dispatches to;
- ``projection_entrypoint`` — the final assemble/render step that turns the
graph into the delivered artifact (e.g. ``clips_to_animatic``);
- ``folder_conventions`` — optional project-folder layout hints.
:class:`nw.Project`, the ``prepare -> plan -> execute`` split, ``stale_after``
freshness, ``nw.jobs`` and the cost gate are all genre-agnostic and serve every
genre unchanged — so *adding a genre is a one-file registration*, the same
open-closed shape as :mod:`nw.renderers` and :mod:`nw.transforms`.
Genres live in the :data:`genres` registry (an :class:`xdol.Registry` with
``on_conflict="error"``). nw ships **no** built-in genres: concrete genres
register themselves from their own packages (``muvid``, ``braidio``) or from
the studio host (``reelee``), which keeps app-layer concerns (output intents,
flavors, prompt packs, cost profiles) out of the substrate. A named preset
*within* a genre (a filled-in default configuration) is a :class:`Template`.
The substrate owns a Template's *identity* (slug/title/description) and carries
a genre-defined ``params`` payload it does **not** interpret — the app that owns
the genre validates and resolves those params (reelee reads ``output_intent`` /
``flavor``; braidio reads a ``format_id``). This keeps the genre *self-describing*
for any consumer (a CLI, an HTTP catalog, an MCP connector) while app-specific
meaning stays in the app.
Naming rationale (Genre vs the alternatives ``kind`` / ``format`` / ...) is in
GitHub issue thorwhalen/nw#10; the "nw owns the engine, each app supplies its
schemas + Transforms" stance is in
``misc/docs/Rendering Provenance and Partial Re-render.md``.
"""
from __future__ import annotations
import inspect
import re
from collections.abc import Mapping
from dataclasses import dataclass, field
from pathlib import Path
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Callable, Optional, Union
from xdol import Registry
if TYPE_CHECKING: # runtime-free: only for the GenreInitializer type hint
from .project import Project
#: Genre lifecycle statuses. ``available`` = usable now; ``experimental`` =
#: usable but unstable; ``planned`` = declared for discovery, not yet ready.
GENRE_STATUSES = ("available", "experimental", "planned")
DFLT_GENRE_STATUS = "available"
def _validate_slug(slug: object, *, what: str) -> None:
"""Assert ``slug`` is a registry-safe key: a non-empty, whitespace-free string.
The one place the slug contract for :class:`Genre`, :class:`Template`, and a
genre resolver is defined — so every registrable thing keyed by a slug fails
fast on a bad key instead of becoming silent dead state.
"""
if not isinstance(slug, str) or not slug.strip():
raise ValueError(f"{what} slug must be a non-empty string")
if any(ch.isspace() for ch in slug):
raise ValueError(f"{what} slug {slug!r} must not contain whitespace")
[docs]
@dataclass(frozen=True)
class Template:
"""A named preset ("subgenre") *within* a genre — a filled-in default config.
AV-general: the substrate owns the Template's identity (``slug``/``title``/
``description``) and carries an opaque ``params`` payload it does **not**
interpret. The app that owns the genre puts meaning in ``params`` (reelee:
``{"output_intent": ..., "flavor": ...}``; braidio: ``{"format_id": ...}``)
and validates/resolves it. Frozen + hashable (``params`` is excluded from
identity and normalized to an immutable mapping), so a Template can live in a
``Genre.templates`` tuple without breaking the genre's ``__hash__``.
>>> t = Template(slug="cinematic_clip", title="Cinematic clip",
... params={"flavor": "fal.cinematic"})
>>> t.params["flavor"]
'fal.cinematic'
>>> t.to_dict()["params"]
{'flavor': 'fal.cinematic'}
"""
slug: str
title: str
description: str = ""
# opaque, genre-defined preset payload — see class docstring. ``compare=False``
# keeps the frozen Template hashable (a Mapping field would break __hash__).
params: Mapping[str, Any] = field(default_factory=dict, compare=False)
def __post_init__(self) -> None:
_validate_slug(self.slug, what="Template")
if not isinstance(self.title, str) or not self.title.strip():
raise ValueError(
f"Template {self.slug!r}: title must be a non-empty string"
)
if not isinstance(self.params, MappingProxyType):
object.__setattr__(self, "params", MappingProxyType(dict(self.params)))
[docs]
def to_dict(self) -> dict:
"""A JSON-able catalog entry: ``{slug, title, description, params}``."""
return {
"slug": self.slug,
"title": self.title,
"description": self.description,
"params": dict(self.params),
}
[docs]
@dataclass(frozen=True)
class Genre:
"""A reusable definition of a *production kind* over the nw substrate.
Pure data: it *references* substrate pieces by name rather than owning
them, so declaring a genre never touches the engine.
>>> slideshow = Genre(
... slug="slideshow",
... title="Slideshow",
... description="Stills over narration, assembled to a video.",
... transform_names=("clips_to_animatic.ffmpeg",),
... projection_entrypoint="clips_to_animatic.ffmpeg",
... )
>>> slideshow.title
'Slideshow'
>>> slideshow.status
'available'
``projection_entrypoint``, when given, must be one of the genre's own
declared transforms or strategies:
>>> Genre(slug="bad", title="Bad", projection_entrypoint="nope")
Traceback (most recent call last):
...
ValueError: Genre 'bad': projection_entrypoint 'nope' is not among its transform_names or strategy_names
"""
slug: str
title: str
description: str = ""
body_schema_uris: tuple[str, ...] = ()
transform_names: tuple[str, ...] = ()
strategy_names: tuple[str, ...] = ()
projection_entrypoint: str | None = None
# ``compare=False`` keeps the descriptor hashable (a dict field would break
# the frozen dataclass's generated ``__hash__``) and keeps folder layout —
# incidental metadata — out of genre *identity*. Normalized to an immutable
# mapping in ``__post_init__`` so a "frozen" Genre is genuinely frozen.
folder_conventions: Mapping[str, str] = field(default_factory=dict, compare=False)
status: str = DFLT_GENRE_STATUS
#: Named presets ("subgenres") within this genre — see :class:`Template`.
templates: tuple[Template, ...] = ()
#: Intake "what are you making?" answers that select this genre (the edge
#: :func:`recommend_genre` walks). App data (e.g. reelee's intake form) owns
#: the vocabulary; the genre just declares which answers it covers.
intake_kinds: tuple[str, ...] = ()
#: A short discriminator slug routing the cost gate to the right estimator
#: (e.g. ``"tts"`` = per-character audio, ``"per_clip"`` = per-render video).
#: The real numbers stay in the app; this is only the routing tag.
cost_profile: str | None = None
#: The "start from scratch" params for this genre (same opaque shape as a
#: :class:`Template`'s ``params``) — used when no template is chosen.
defaults: Mapping[str, Any] = field(default_factory=dict, compare=False)
def __post_init__(self) -> None:
# Normalize every sequence field to a tuple up front — symmetrically with
# the Mapping-field normalization below — so a "frozen" Genre built with a
# list (e.g. ``templates=[Template(...) for ...]``) is genuinely immutable
# AND hashable, rather than a mutable list silently breaking ``__hash__``
# only at the first set/dict-key use far from here.
for _name in (
"body_schema_uris",
"transform_names",
"strategy_names",
"templates",
"intake_kinds",
):
_value = getattr(self, _name)
if not isinstance(_value, tuple):
object.__setattr__(self, _name, tuple(_value))
_validate_slug(self.slug, what="Genre")
if not isinstance(self.title, str) or not self.title.strip():
raise ValueError(f"Genre {self.slug!r}: title must be a non-empty string")
if self.status not in GENRE_STATUSES:
raise ValueError(
f"Genre {self.slug!r}: status {self.status!r} not in {GENRE_STATUSES}"
)
pe = self.projection_entrypoint
if (
pe is not None
and pe not in self.transform_names
and pe not in self.strategy_names
):
raise ValueError(
f"Genre {self.slug!r}: projection_entrypoint {pe!r} is not among "
"its transform_names or strategy_names"
)
if not isinstance(self.folder_conventions, MappingProxyType):
object.__setattr__(
self,
"folder_conventions",
MappingProxyType(dict(self.folder_conventions)),
)
if not all(isinstance(t, Template) for t in self.templates):
raise ValueError(f"Genre {self.slug!r}: templates must all be nw.Template")
slugs = [t.slug for t in self.templates]
if len(slugs) != len(set(slugs)):
raise ValueError(
f"Genre {self.slug!r}: template slugs must be unique, got {slugs}"
)
if any((not isinstance(k, str) or not k.strip()) for k in self.intake_kinds):
raise ValueError(
f"Genre {self.slug!r}: intake_kinds must be non-empty strings"
)
if self.cost_profile is not None and (
not isinstance(self.cost_profile, str) or not self.cost_profile.strip()
):
raise ValueError(
f"Genre {self.slug!r}: cost_profile must be a non-empty string or None"
)
if not isinstance(self.defaults, MappingProxyType):
object.__setattr__(self, "defaults", MappingProxyType(dict(self.defaults)))
[docs]
def template(self, slug: str) -> Template:
"""Look up one of this genre's :class:`Template`\\ s by slug (KeyError if absent)."""
for candidate in self.templates:
if candidate.slug == slug:
return candidate
known = [t.slug for t in self.templates]
raise KeyError(f"Genre {self.slug!r} has no template {slug!r}; has: {known}")
[docs]
def list_templates(self) -> list[str]:
"""This genre's Template slugs, in declared order."""
return [t.slug for t in self.templates]
[docs]
def to_dict(self) -> dict:
"""A JSON-able catalog entry — the shape apps serve to a frontend / MCP client.
Templates are emitted with their opaque ``params`` (not flattened), and
``intake_kinds``/``cost_profile``/``defaults`` ride at the genre level, so a
consumer needs no app-specific knowledge to render the catalog.
"""
return {
"slug": self.slug,
"title": self.title,
"description": self.description,
"status": self.status,
"ready": self.is_ready(),
"intake_kinds": list(self.intake_kinds),
"cost_profile": self.cost_profile,
"defaults": dict(self.defaults),
"templates": [t.to_dict() for t in self.templates],
}
[docs]
def missing_strategies(self) -> list[str]:
"""Declared ``strategy_names`` not (yet) present in ``nw.renderers``."""
from .renderers import strategies as _strategies
return [n for n in self.strategy_names if n not in _strategies]
[docs]
def is_ready(self) -> bool:
"""True iff every declared transform and strategy is registered.
A ``planned`` genre may legitimately be *not* ready; an ``available``
one that isn't ready is a wiring bug worth catching in a test.
"""
return not self.missing_transforms() and not self.missing_strategies()
#: The genre registry. ``on_conflict="error"`` so a misconfigured plugin can't
#: silently shadow another package's genre.
genres: Registry = Registry(name="nw.genres", on_conflict="error")
"""Public registry of :class:`Genre` instances, keyed by ``genre.slug``.
Apps add genres via :func:`register_genre` (or ``nw.register_genre``)."""
[docs]
def register_genre(genre: Genre) -> Genre:
"""Register a :class:`Genre` under its ``slug``; returns it for inline use.
>>> g = register_genre(Genre(slug="doctest_demo", title="Demo"))
>>> get_genre("doctest_demo").title
'Demo'
>>> "doctest_demo" in list_genres()
True
>>> del genres["doctest_demo"] # keep the shared registry clean
"""
if not isinstance(genre, Genre):
raise TypeError(f"register_genre expects a Genre, got {type(genre).__name__}")
genres.register(genre.slug, genre)
return genre
[docs]
def get_genre(slug: str) -> Genre:
"""Look up a genre by slug; raises :class:`KeyError` with the known slugs."""
if slug not in genres:
known = sorted(genres.keys())
raise KeyError(
f"No genre {slug!r}; registered: {known}. Apps register genres via "
"`nw.register_genre(Genre(...))`."
)
return genres[slug]
[docs]
def list_genres() -> list[str]:
"""Return all registered genre slugs (sorted)."""
return sorted(genres.keys())
[docs]
def genre_catalog() -> list[dict]:
"""Every registered genre as a JSON-able catalog entry (sorted by slug).
This is the generic, app-agnostic catalog an HTTP route / MCP tool serves; see
:meth:`Genre.to_dict` for the entry shape.
"""
return [get_genre(slug).to_dict() for slug in list_genres()]
[docs]
def describe_genre(slug: str) -> dict:
"""One genre's catalog entry (raises :class:`KeyError` if the slug is unknown)."""
return get_genre(slug).to_dict()
[docs]
def recommend_genre(kind: str | None) -> str | None:
"""The slug of the genre whose ``intake_kinds`` contains ``kind`` (first in slug
order), or ``None`` when ``kind`` is falsy / unmatched.
>>> g = register_genre(Genre(slug="_rec_demo", title="Rec", intake_kinds=("essay",)))
>>> recommend_genre("essay")
'_rec_demo'
>>> recommend_genre("nope") is None and recommend_genre(None) is None
True
>>> del genres["_rec_demo"]
"""
if not kind:
return None
for slug in list_genres():
if kind in get_genre(slug).intake_kinds:
return slug
return None
[docs]
def resolve_defaults(genre: str, template: str | None = None) -> dict:
"""Resolve a genre (+ optional template) to the params for a new project.
Returns ``{"genre": slug, "template": template_or_None, "params": {...}}`` — the
chosen :class:`Template`'s ``params`` when ``template`` is given, else the genre's
``defaults``. Raises :class:`KeyError` on an unknown genre or template. The caller
(app) interprets ``params`` (reelee reads ``output_intent``/``flavor``; braidio a
``format_id``).
"""
g = get_genre(genre)
params = dict(g.defaults) if template is None else dict(g.template(template).params)
return {"genre": genre, "template": template, "params": params}
#: A genre resolver: ``(genre, template) -> params`` — an owning-app-defined,
#: JSON-able ``params`` payload (e.g. reelee's ``{output_intent, flavor}``, braidio's
#: ``{format_id}``) for the chosen template (or ``None`` = "start from scratch").
#: :func:`resolve_genre` wraps this in the standard ``{genre, template, params}``
#: envelope — the resolver returns ONLY the bare params. Registered per genre via
#: :func:`register_genre_resolver`.
GenreResolver = Callable[[Genre, Optional[str]], dict]
#: Registry of per-genre resolvers, keyed by genre slug. A genre's **owning app**
#: registers how to turn a ``(genre, template)`` into its params, so a host that
#: aggregates many genres (a CLI, an HTTP API, an MCP connector) can create ANY
#: genre's project via :func:`resolve_genre` without hardcoding which app owns it.
genre_resolvers: Registry = Registry(name="nw.genre_resolvers", on_conflict="error")
[docs]
def register_genre_resolver(slug: str, resolver: "GenreResolver") -> "GenreResolver":
"""Register a resolver for a genre slug; returns it for inline use.
Called by the genre's owning app. ``resolver(genre, template) -> params`` maps a
chosen template (or ``None`` for "start from scratch") to that app's bare params
payload; :func:`resolve_genre` adds the ``{genre, template, params}`` envelope.
Independent of genre *registration order* (keyed by the slug string).
>>> _ = register_genre(Genre(slug="_resolver_demo", title="Demo",
... defaults={"look": "plain"}))
>>> _ = register_genre_resolver("_resolver_demo",
... lambda genre, template: {"look": genre.defaults["look"], "via": "resolver"})
>>> resolve_genre("_resolver_demo")
{'genre': '_resolver_demo', 'template': None, 'params': {'look': 'plain', 'via': 'resolver'}}
>>> del genres["_resolver_demo"]; del genre_resolvers["_resolver_demo"]
"""
_validate_slug(slug, what="genre resolver")
if not callable(resolver):
raise TypeError(
f"register_genre_resolver expects a callable, got {type(resolver).__name__}"
)
genre_resolvers.register(slug, resolver)
return resolver
[docs]
def resolve_genre(genre: str, template: Optional[str] = None) -> dict:
"""Resolve a genre (+ optional template) to the standard creation envelope.
Always returns ``{"genre": slug, "template": template_or_None, "params": {...}}`` —
ONE stable contract for every host, regardless of whether the genre has a resolver.
``params`` comes from the genre's registered resolver
(:func:`register_genre_resolver`) when one exists, else from the generic
:func:`resolve_defaults` (the template's params, or the genre's ``defaults``).
Raises :class:`KeyError` on an unknown genre or — **uniformly, resolver or not** —
an unknown ``template`` slug (the substrate owns template identity; a resolver only
interprets params, it doesn't get to invent template slugs).
>>> _ = register_genre(Genre(slug="_rg_demo", title="Demo",
... defaults={"flavor": "cinematic"}))
>>> resolve_genre("_rg_demo") # no resolver registered -> generic params
{'genre': '_rg_demo', 'template': None, 'params': {'flavor': 'cinematic'}}
>>> del genres["_rg_demo"]
"""
g = get_genre(genre)
chosen = (
g.template(template) if template is not None else None
) # validates the slug
if genre in genre_resolvers:
params = genre_resolvers[genre](g, template)
else:
params = dict(g.defaults) if chosen is None else dict(chosen.params)
return {"genre": genre, "template": template, "params": params}
# ---------------------------------------------------------------------------
# Genre INITIALIZERS — the *apply* half of the genre→project contract.
#
# :func:`resolve_genre` is the *pure* half: ``(genre, template) -> params`` (the
# JSON-able creation envelope). A genre initializer is its *side-effecting* twin:
# ``(genre, template, project, params) -> None`` — it **seeds a freshly-created
# project** for the chosen genre/template (reelee writes an output-intent
# annotation; braidio applies its format at *render* time, so it registers none).
# Together they let a host that aggregates many genres (a CLI, an HTTP API, an MCP
# connector) create ANY genre's project — resolve to get the params, create the
# bare project, then initialize — without hardcoding which app owns the genre.
# ---------------------------------------------------------------------------
#: A genre initializer: ``(genre, template, project, params) -> None``. The
#: side-effecting apply-counterpart to a :data:`GenreResolver`'s pure
#: ``(genre, template) -> params``. It receives the full resolver context (the
#: :class:`Genre` object + chosen ``template`` slug) plus the freshly-created
#: ``project`` to seed and the resolved ``params``. reelee's writes the
#: output-intent annotation; a genre that seeds nothing on create registers none.
#:
#: **Contract:** an initializer MUST confine all side effects to ``project``
#: (typically writing annotations under ``project.root``) — so a host can revert a
#: failed create by deleting the project folder — or own its own rollback. Writing
#: to a shared store / external service / global registry breaks that guarantee.
GenreInitializer = Callable[
["Genre", Optional[str], "Project", Mapping[str, Any]], None
]
#: Registry of per-genre initializers, keyed by genre slug. A genre's **owning
#: app** registers how to seed a fresh project for that genre; a genre that seeds
#: nothing on create simply registers none (:func:`initialize_genre` no-ops).
genre_initializers: Registry = Registry(
name="nw.genre_initializers", on_conflict="error"
)
[docs]
def register_genre_initializer(
slug: str, initializer: "GenreInitializer"
) -> "GenreInitializer":
"""Register an initializer for a genre slug; returns it for inline use.
Called by the genre's owning app. ``initializer(genre, template, project,
params) -> None`` seeds a freshly-created project for the chosen genre/template
(see :data:`GenreInitializer` for the side-effect contract);
:func:`initialize_genre` dispatches to it. Independent of genre *registration
order* (keyed by the slug string). A genre that seeds nothing on create needs
no initializer at all.
>>> _ = register_genre(Genre(slug="_init_demo", title="Demo",
... defaults={"look": "plain"}))
>>> seen = {}
>>> def _seed(genre, template, project, params):
... seen["applied"] = (genre.slug, template, params)
>>> _ = register_genre_initializer("_init_demo", _seed)
>>> initialize_genre("_init_demo", object()) # params default to the genre's
>>> seen["applied"]
('_init_demo', None, {'look': 'plain'})
>>> del genres["_init_demo"]; del genre_initializers["_init_demo"]
"""
_validate_slug(slug, what="genre initializer")
if not callable(initializer):
raise TypeError(
"register_genre_initializer expects a callable, got "
f"{type(initializer).__name__}"
)
genre_initializers.register(slug, initializer)
return initializer
[docs]
def initialize_genre(
genre: str,
project: "Project",
*,
template: Optional[str] = None,
params: Optional[Mapping[str, Any]] = None,
) -> None:
"""Seed a freshly-created ``project`` for ``genre`` (+ optional ``template``).
The side-effecting apply-counterpart to :func:`resolve_genre`. Dispatches to
the genre's registered initializer (:func:`register_genre_initializer`) when
one exists; **when none is registered this is a no-op** — the correct default
for a genre that seeds nothing on create (e.g. one whose preset is applied at
render time).
``params`` is the resolved creation params (from :func:`resolve_genre`); when
``None`` it is resolved from the genre's ``template``/``defaults`` here, so
``initialize_genre(genre, project)`` seeds a project in the genre's defaults in
one call. Raises :class:`KeyError` on an unknown genre or — **uniformly** — an
unknown ``template`` slug (matching :func:`resolve_genre`).
**The envelope is persisted** (nw#32): after the initializer succeeds (or
no-ops), the resolved ``{genre, template, params}`` is recorded on the
project's graph — so "what genre is this project?" stays answerable after
this call returns, on the host-creates path exactly as on
:func:`create_genre_project`'s. Written *after* the seed on purpose: a
recorded envelope certifies a completed initialization, never a failed
one. A ``project`` stand-in without a ``graph`` (a test double, a
factory that returns no live project) skips the recording; read it back
via :meth:`nw.Project.resolved_genre`.
>>> _ = register_genre(Genre(slug="_noinit_demo", title="Demo"))
>>> initialize_genre("_noinit_demo", object()) # no initializer -> no seed
>>> del genres["_noinit_demo"]
"""
g = get_genre(genre) # validates the genre slug (KeyError otherwise)
if params is None:
params = resolve_genre(genre, template)["params"] # also validates template
elif template is not None:
g.template(template) # validate template uniformly even when params given
if genre in genre_initializers:
genre_initializers[genre](g, template, project, dict(params))
_persist_genre_envelope(project, genre=genre, template=template, params=params)
def _persist_genre_envelope(
project,
*,
genre: str,
template: Optional[str],
params: Mapping[str, Any],
) -> bool:
"""Record the resolved envelope on ``project``'s graph; return whether it was.
Duck-typed on ``project.graph.set_genre_envelope`` because the genre
machinery legitimately runs against non-:class:`nw.Project` objects (a
factory that returns ``{"project": None}``, test doubles). A real nw
project always records; anything else is the caller's to record.
"""
graph = getattr(project, "graph", None)
set_envelope = getattr(graph, "set_genre_envelope", None)
if not callable(set_envelope):
return False
from .bodies import GenreEnvelopeBodyV1
set_envelope(
GenreEnvelopeBodyV1(genre=genre, template=template, params=dict(params))
)
return True
# ---------------------------------------------------------------------------
# Genre PROJECT FACTORIES — the *create* half of the genre→project contract, for
# PLUGGED-IN genres (ones a host aggregates but does not natively own).
#
# resolve_genre (params) + initialize_genre (seed) assume the host already HAS a
# project to seed. But when a host (e.g. an MCP connector) aggregates a genre owned by
# ANOTHER app — reelee hosting braidio's ``commentary_weave`` — it can't build that
# genre's project itself (a braidio project lives in the owning app's per-user
# workspace, not a reelee sibling folder). A project factory lets the owning app supply
# "create a fresh project for this genre in the CALLER's own space", so a host can offer
# ``create_project(genre)`` for any plugged-in genre without knowing its storage. A
# host's OWN genres (which it places itself) do NOT use this registry.
#
# PLACEMENT (``projects_dir``). The original contract had no location argument, so the
# genre's app necessarily decided storage — sound while a host only aggregated each
# genre's *tools* in each genre's own workspace, and wrong the moment a host has to
# *serve* the project: a project created under the guest app's data home is a sibling of
# nothing the host can address, so the host's lister never lists it and its project
# header cannot name it. The rule, which generalises past any one genre:
#
# a genre project factory places a project where its caller asks;
# it does not own the location.
#
# So the contract gains ``projects_dir``, defaulting to ``None`` = the owning app's own
# workspace — every pre-existing caller and factory keeps working unchanged.
# ---------------------------------------------------------------------------
#: The keyword a factory declares to accept host placement — see
#: :data:`GenreProjectFactory` and :func:`can_place_genre_project`.
PLACEMENT_ARG = "projects_dir"
#: A genre project factory:
#: ``(caller, project_id, *, title, template, params, projects_dir=None) -> dict``.
#: Creates a NEW project for a plugged-in ``genre`` and returns info including
#: ``{"project": <created project>, ...}`` — see :func:`create_genre_project` (which
#: seeds it then strips the live object from the JSON result).
#:
#: **Placement.** ``projects_dir`` is the directory the new project folder is created
#: *in* — the new project's root is ``projects_dir/<project_id>``, so it is a sibling of
#: whatever else the host keeps there. ``None`` (the default, and what a host that has no
#: opinion passes) means *the owning app's own workspace for* ``caller``, which is the
#: pre-placement behaviour. A factory that accepts ``projects_dir`` MUST honour it;
#: :func:`create_genre_project` verifies the created project actually landed there and
#: rolls back if it did not, because a factory that quietly places it elsewhere is the
#: exact failure placement exists to remove.
#:
#: **Caller-space contract (unchanged for** ``projects_dir=None`` **).** With no
#: placement, confine creation to the caller's own space so a host can offer it
#: multi-tenant safely. With a placement, the *host* has already decided the caller's
#: space and owns that guarantee. A host-tenancy genre does NOT register here.
GenreProjectFactory = Callable[..., dict]
#: Registry of per-genre project factories, keyed by genre slug (the owning app registers).
genre_project_factories: Registry = Registry(
name="nw.genre_project_factories", on_conflict="error"
)
[docs]
def register_genre_project_factory(
slug: str, factory: "GenreProjectFactory"
) -> "GenreProjectFactory":
"""Register a project factory for a genre slug; returns it for inline use.
Called by the genre's **owning app** so a host that aggregates the genre can create
its projects via :func:`create_genre_project` without knowing its storage. See
:data:`GenreProjectFactory` for the signature + the caller-space contract.
>>> _ = register_genre(Genre(slug="_pf_demo", title="Demo",
... defaults={"format_id": "solo"}))
>>> made = {}
>>> def _f(caller, project_id, *, title, template, params):
... made.update(caller=caller, project_id=project_id, params=params)
... return {"project": None, "project_id": project_id}
>>> _ = register_genre_project_factory("_pf_demo", _f)
>>> create_genre_project("_pf_demo", "u@x.com", "p1")["project_id"]
'p1'
>>> (made["caller"], made["params"])
('u@x.com', {'format_id': 'solo'})
>>> del genres["_pf_demo"]; del genre_project_factories["_pf_demo"]
"""
_validate_slug(slug, what="genre project factory")
if not callable(factory):
raise TypeError(
"register_genre_project_factory expects a callable, got "
f"{type(factory).__name__}"
)
genre_project_factories.register(slug, factory)
return factory
[docs]
def has_genre_project_factory(slug: str) -> bool:
"""True iff a plugged-in project factory is registered for ``slug``."""
return slug in genre_project_factories
def _accepts_placement(factory: "GenreProjectFactory") -> bool:
"""Whether ``factory`` can actually be *called* with :data:`PLACEMENT_ARG`.
Signature inspection rather than a required argument + a major version bump,
because this registry holds **third-party callables**: nw is published, and the
apps that register here (braidio, muvid, reelee) release on their own cadences,
so requiring the argument would turn every not-yet-updated factory into a
``TypeError`` on the create path — the most user-visible one there is. Adaptation
costs nothing when no placement is asked for, and :func:`create_genre_project`
refuses (rather than silently misplacing) when one is.
The probe is a **bind of the whole call**, not a name lookup: ``projects_dir``
declared positional-only, or as ``*projects_dir``, is a name in ``parameters``
that cannot be passed as a keyword, and a membership test reads all three alike.
``bind_partial`` is not enough either — it tolerates a positional-only parameter
passed by keyword, which the real call does not. So the probe binds exactly the
argument list :func:`create_genre_project` is about to pass; the values are
irrelevant, only whether they go anywhere.
``**kwargs`` binds, so it counts as accepting: it is the only honest reading of
the signature. It is also why acceptance is not the guarantee — the *outcome*
check in :func:`create_genre_project` is, and it catches a factory that takes the
argument and ignores it however the signature is spelled.
"""
try:
sig = inspect.signature(factory)
except (TypeError, ValueError): # a C callable with no introspectable signature
return False
try:
sig.bind(
None, # caller
None, # project_id
title=None,
template=None,
params={},
**{PLACEMENT_ARG: None},
)
except TypeError:
return False
return True
[docs]
def can_place_genre_project(slug: str) -> bool:
"""True iff ``slug``'s registered factory accepts host **placement**.
The question a host asks *before* offering "create a project of this genre here":
a genre whose factory predates :data:`PLACEMENT_ARG` can still be created, but
only in its own app's workspace — where the host cannot address it. False for an
unregistered genre.
>>> def _old(caller, project_id, *, title, template, params):
... return {"project": None}
>>> def _new(caller, project_id, *, title, template, params, projects_dir=None):
... return {"project": None}
>>> _ = register_genre_project_factory("_place_old", _old)
>>> _ = register_genre_project_factory("_place_new", _new)
>>> can_place_genre_project("_place_old"), can_place_genre_project("_place_new")
(False, True)
>>> can_place_genre_project("_place_nope")
False
>>> del genre_project_factories["_place_old"]
>>> del genre_project_factories["_place_new"]
"""
if slug not in genre_project_factories:
return False
return _accepts_placement(genre_project_factories[slug])
def _project_root(project) -> Optional[Path]:
"""``project.root`` as a :class:`Path`, or ``None`` when there isn't one.
Tolerates a missing attribute, ``None``, a ``str``, and a property that raises —
the last because this is called from rollback paths, where an exception would
replace the real error with a confusing one from inside the cleanup.
"""
try:
root = getattr(project, "root", None)
except Exception:
return None
if root is None:
return None
try:
return Path(root)
except TypeError:
return None
def _verify_placement(project, projects_dir: Path, *, genre: str, project_id: str):
"""Assert the just-created ``project`` IS ``projects_dir/<project_id>``.
The mechanism that makes placement a contract rather than a hint. A factory may
declare the keyword and ignore it (or absorb it into ``**kwargs``), and the
resulting project is *fine on disk* — it is just somewhere the host cannot see,
which is indistinguishable from "the create silently did nothing" on every host
surface. So the outcome is checked, not the intention.
**The whole path, not a suffix of it.** The realistic misplacement is not an
exotic one: it is the same tail under a different data root
(``{braidio_home}/projects/{email}/`` against ``{reelee_home}/projects/{email}/``),
which is exactly what a comparison on the last component alone would wave through.
**The basename too**, because the host does not get the created root back — the
result is JSON-able and deliberately drops the live project — so it addresses the
new project as ``projects_dir/<project_id>``. Verifying only the parent would
leave the half the host actually relies on unchecked; a factory that slugifies or
prefixes the id produces a project nothing can then open by name.
**An unverifiable outcome is a failure, not a pass.** A factory that accepts the
placement and returns no live project (or one with no ``root``) leaves nw unable
to say where the project went, and the direction that silence resolves to must
not be "success" — that is precisely the two-worlds report this exists to
prevent. A factory that predates the argument is never asked, so nothing that
worked before is affected.
``.resolve()`` on both sides is load-bearing, not tidiness: ``nw.Project.__init__``
resolves its root, so a factory built on it returns a *resolved* path while a
host's ``projects_dir`` is typically not resolved — under a symlinked data root a
non-resolving comparison refuses a correctly-placed project.
"""
want = Path(projects_dir).resolve()
got = _project_root(project)
if got is None:
raise RuntimeError(
f"genre {genre!r}'s project factory accepted {PLACEMENT_ARG!r} but "
"returned no created project, so nw cannot verify where the project "
f"went. A factory that accepts a placement must return "
'``{"project": <the created project>, ...}``.'
)
got = got.resolve()
if got != want / project_id:
raise RuntimeError(
f"genre {genre!r}'s project factory did not honour projects_dir: asked "
f"for {want / project_id}, got {got}. A factory that declares "
f"{PLACEMENT_ARG!r} must create at {PLACEMENT_ARG}/<project_id> — the "
"host addresses the new project by exactly that path, and a project "
"anywhere else is unreachable by whoever asked for it."
)
[docs]
def create_genre_project(
genre: str,
caller: str,
project_id: str,
*,
title: Optional[str] = None,
template: Optional[str] = None,
projects_dir: Optional[Union[str, Path]] = None,
) -> dict:
"""Create + seed a new project for a PLUGGED-IN ``genre`` in ``caller``'s space.
The *create*-counterpart to :func:`resolve_genre` (params) + :func:`initialize_genre`
(seed): resolve → the genre's factory (create in the caller's space) → initialize.
**All-or-nothing** — if seeding fails the just-created project is rolled back (its
folder removed) and the error re-raised. Returns the factory's JSON-able info (minus
the live ``project``) plus the resolved ``{genre, template, params}`` envelope, so the
caller can immediately address the new project (e.g. by ``project_id``). The same
envelope is **persisted on the project** by the initialize step (nw#32) — under the
all-or-nothing guarantee, since a failure there rolls the whole create back — so the
association survives this call returning; read it back via
:meth:`nw.Project.resolved_genre`.
``projects_dir`` is the **placement**: the directory to create the project folder
in, so a host that will *serve* the project can put it where its own resolver
looks. ``None`` (the default) leaves placement to the genre's app, which is the
pre-placement behaviour. Ask :func:`can_place_genre_project` first, or handle the
``TypeError`` a pre-placement factory raises here — the request is refused **before
any filesystem effect**, never quietly satisfied somewhere else.
Raises :class:`KeyError` on an unknown genre/template, or a genre with no registered
factory (a host's own genre is created by the host, not via this path);
:class:`TypeError` when ``projects_dir`` is given for a factory that does not accept
it; :class:`RuntimeError` (after rolling the create back) when a factory accepted a
placement and did not honour it.
"""
if genre not in genre_project_factories:
known = sorted(genre_project_factories.keys())
raise KeyError(
f"genre {genre!r} has no project factory (registered: {known}); a host's "
"own genres are created by the host, not via create_genre_project."
)
factory = genre_project_factories[genre]
extra: dict = {}
placement: Optional[Path] = None
if projects_dir is not None:
placement = Path(projects_dir)
if not str(projects_dir).strip():
# "" is not "no opinion": `Path("")` is the process's CWD, so accepting a
# blank would create real user projects wherever the server happens to be
# running — typically the deploy tree. An empty placement is a caller bug
# (a tool schema defaulting to ""), and a loud one beats a silent one.
raise ValueError(
f"{PLACEMENT_ARG} is empty; pass a directory, or None for the "
"genre's own workspace."
)
# Refuse BEFORE resolving or creating anything. Ignoring the placement would
# succeed, report success, and leave the project where the asking host cannot
# see it — the failure this argument exists to remove.
if not _accepts_placement(factory):
raise TypeError(
f"genre {genre!r}'s project factory does not accept "
f"{PLACEMENT_ARG!r}, so it cannot create a project at "
f"{projects_dir}; it places projects in its own app's workspace. "
"Ask can_place_genre_project() first, or update that factory to "
f"accept {PLACEMENT_ARG}."
)
extra[PLACEMENT_ARG] = placement
envelope = resolve_genre(genre, template) # validates genre + template
params = envelope["params"]
info = factory(
caller,
project_id,
title=title or project_id,
template=template,
params=params,
**extra,
)
project = info.get("project") if isinstance(info, dict) else None
try:
if placement is not None:
_verify_placement(project, placement, genre=genre, project_id=project_id)
initialize_genre(genre, project, template=template, params=params)
except Exception:
# all-or-nothing: leave no half-built orphan — but see _rollback_project on
# why a placement BOUNDS what may be deleted rather than widening it.
_rollback_project(project, within=placement)
raise
result = (
{k: v for k, v in info.items() if k != "project"}
if isinstance(info, dict)
else {}
)
result["genre"] = genre
result["template"] = template
result["params"] = params
return result
def _rollback_project(project, *, within: Optional[Path] = None) -> None:
"""Best-effort delete of a just-created project after a failure (all-or-nothing).
Removes ``project.root`` if present. Without a placement that is as safe as it
was before placement existed: the factory created in its own app's space, and
``project.root`` is what it just made there.
``within`` is the load-bearing part. A placement adds a second, much easier
trigger for this function — :func:`_verify_placement` — and that trigger fires
*precisely when nw has concluded it does not know what the factory did*.
Recursively deleting a path nw does not understand, inside the **host's** tree
rather than the guest app's, is not a rollback: a factory that places correctly
on disk and returns the wrong ``root`` (its parent, say, or ``Path.home()``)
would have the host's whole per-caller projects directory removed, reported as a
clean all-or-nothing abort.
So when a placement was asked for, nothing outside it is touched: only a strict
descendant of ``within`` is deleted, which is both what nw asked the factory to
create and the only thing it can know was not already there. Anything else is
left on disk, and the error the caller sees names where it is.
"""
root = _project_root(project)
if root is None:
return
if within is not None:
try:
resolved, bound = root.resolve(), Path(within).resolve()
except OSError:
return
if bound not in resolved.parents:
return # outside the placement — not ours to delete
import shutil
shutil.rmtree(root, ignore_errors=True)
# ---------------------------------------------------------------------------
# Genre operations — what a host SERVES on a genre's project, without importing it
#
# A project factory lets a host *create* a guest genre's project; an op registry lets
# it *work on* one. A genre registers its project operations here as plain functions
# ``(project, **params) -> dict`` (JSON-able in and out), each with a plain-language
# title, what it does to the project (``effect``) and whether it is slow enough that
# the host must run it in the background (``runs``). A host reads this registry to
# build every surface it offers — an HTTP route, a frontend command, an assistant tool —
# from the one list, instead of re-listing a genre's operations by hand (a list written
# twice is two implementations; reelee's HTTP<->MCP parity test measured twelve drifts).
#
# Deliberately tiny: no base class, no plugin loader, no dispatch. nw owns the
# *description* of an op and the check that its parameters are JSON; what a refusal
# looks like on the wire, how a job is queued and who may call what are the host's.
# ---------------------------------------------------------------------------
#: What an op does to the project. ``destroy`` = it removes (or invalidates) something
#: the user made or uploaded — a host gates these behind a confirmation.
GENRE_OP_EFFECTS = ("read", "write", "render", "destroy")
#: How a host should run an op: ``now`` = answer in the request; ``job`` = slow (seconds
#: to minutes), so run it in the background and let the caller watch it.
GENRE_OP_RUNS = ("now", "job")
_OP_NAME_PATTERN = re.compile(r"^[a-z][a-z0-9_]*$")
[docs]
class GenreOpRefused(ValueError):
"""An op DELIBERATELY declined — no song yet, an unknown clip, an edit that does not
hold. The base a genre derives its refusal type from (muvid's ``FootageError``).
A host maps exactly this to a client-facing refusal (reelee: ``422``); any other
exception out of an op — a plain ``ValueError`` included — is a bug, reported as
one (``500`` with the traceback logged). A ``ValueError`` so code already catching
that keeps working.
"""
[docs]
class GenreOpCancelled(Exception):
"""An op stopped because the host asked it to (its ``should_cancel`` returned True).
Not a refusal and not a failure: a host that cancelled a job records it as
cancelled. Raised by the op, between steps, when it was given a
:data:`CANCEL_PARAM` host parameter that says stop.
"""
#: The host-parameter name for cancellation: an op that lists it in ``host_params``
#: receives a zero-argument callable, polls it between steps, and raises
#: :class:`GenreOpCancelled` when it returns True.
CANCEL_PARAM = "should_cancel"
[docs]
class UnknownGenreOpError(KeyError):
"""``genre_op`` was asked for a name the genre does not register.
A :class:`KeyError` so a caller that already catches unknown-key lookups keeps
working; its message names the known ops, because "no such op" with no menu is a
dead end for a model choosing among them.
"""
def __str__(self) -> str: # KeyError's default repr-quotes the whole message
return str(self.args[0]) if self.args else ""
def _op_params_model(fn: Callable, *, name: str, host_params: tuple = ()):
"""The pydantic model of ``fn``'s CLIENT parameters: those after the first (the
project), minus ``host_params`` (which the host supplies and a client never may).
Refuses — at registration, not at call time — anything a JSON caller could not
satisfy: ``*args``/``**kwargs``, a positional-only parameter, a missing annotation,
or a type pydantic cannot express as JSON Schema. ``extra="forbid"`` so an unknown
parameter is an error rather than silently dropped.
"""
from pydantic import ConfigDict, create_model
try:
sig = inspect.signature(fn, eval_str=True)
except (TypeError, ValueError, NameError) as exc:
raise TypeError(
f"genre op {name!r}: cannot read its signature ({exc})"
) from exc
params = list(sig.parameters.values())
if not params or params[0].kind not in (
inspect.Parameter.POSITIONAL_ONLY,
inspect.Parameter.POSITIONAL_OR_KEYWORD,
):
raise TypeError(
f"genre op {name!r}: fn must take the project as its first positional "
"parameter"
)
by_name = {p.name: p for p in params[1:]}
for h in host_params:
p = by_name.get(h)
if p is None or p.kind not in (
inspect.Parameter.POSITIONAL_OR_KEYWORD,
inspect.Parameter.KEYWORD_ONLY,
):
raise TypeError(
f"genre op {name!r}: host parameter {h!r} is not a keyword parameter "
"of fn, so the host could not pass it"
)
fields: dict = {}
for p in params[1:]:
if p.name in host_params:
continue
if p.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD):
raise TypeError(
f"genre op {name!r}: *{p.name}/**{p.name} cannot be described as JSON "
"parameters; declare each parameter"
)
if p.kind is inspect.Parameter.POSITIONAL_ONLY:
raise TypeError(
f"genre op {name!r}: parameter {p.name!r} is positional-only; ops are "
"called with keyword parameters"
)
if p.annotation is inspect.Parameter.empty:
raise TypeError(
f"genre op {name!r}: parameter {p.name!r} has no annotation, so no "
"JSON Schema can be derived for it"
)
default = ... if p.default is inspect.Parameter.empty else p.default
fields[p.name] = (p.annotation, default)
model_name = "".join(part.title() for part in name.split("_")) + "Params"
try:
model = create_model(
model_name, __config__=ConfigDict(extra="forbid"), **fields
)
model.model_json_schema() # a non-JSON type fails here, loudly
except Exception as exc: # pydantic raises several types; all mean "not JSON"
raise TypeError(
f"genre op {name!r}: its parameters are not JSON-describable ({exc})"
) from exc
return model
[docs]
@dataclass(frozen=True)
class GenreOp:
"""One operation a genre offers on its projects — a row a host builds surfaces from.
``fn(project, **params) -> dict`` does the work; ``params`` and the result are
JSON-able. ``title`` is a short plain-language imperative ("Find where each video
fits") — it becomes a button or command title. ``description`` is model-facing and
defaults to ``fn``'s docstring. ``effect`` is one of :data:`GENRE_OP_EFFECTS`,
``runs`` one of :data:`GENRE_OP_RUNS`.
Construction validates everything, including that every parameter after the
project is annotated with a JSON-describable type — a bad op fails where it is
declared, not in front of a user.
``host_params`` names parameters the HOST supplies, never a client: a streamed
upload's temporary ``path``, its original ``filename``. They are left out of
:attr:`params_schema` (so a client that sends one fails validation — the schema is
``additionalProperties: false``), they must be keyword parameters of ``fn``, and
:meth:`run` passes them through from its ``host`` argument without validating them
— the host produced them. A host reads ``host_params`` in :meth:`to_dict` to know
which ops take an upload. This is the boundary that stops a generic op route from
letting a client name a file on the server.
Two host parameters have agreed meanings: an upload's ``path`` (with
``max_upload_bytes``, the op's own ceiling the host enforces WHILE streaming, before
the op ever runs) and :data:`CANCEL_PARAM` (a zero-argument callable the op polls,
raising :class:`GenreOpCancelled`). A deliberate refusal is a
:class:`GenreOpRefused`.
``spends`` says the op MAY spend money (it can reach a paid API). The federation's
rule is that an unknown cost forces approval, so a host must route a ``spends`` op
through its money approval — or not offer it. ``False`` (the default) is a claim the
genre makes: nothing this op calls bills anyone.
>>> def _rename(project, *, title: str, loud: bool = False) -> dict:
... '''Rename the project.'''
... return {"title": title.upper() if loud else title}
>>> op = GenreOp("rename", _rename, title="Rename it")
>>> op.description, op.effect, op.runs
('Rename the project.', 'write', 'now')
>>> sorted(op.params_schema["properties"]), op.params_schema["required"]
(['loud', 'title'], ['title'])
>>> op.run(None, {"title": "x", "loud": True})
{'title': 'X'}
An op taking an upload, whose ``path`` only the host may give:
>>> def _ingest(project, *, path: str, name: str = "") -> dict:
... return {"path": path, "name": name}
>>> up = GenreOp("ingest", _ingest, title="Add a file", host_params=("path",))
>>> list(up.params_schema["properties"]), up.to_dict()["host_params"]
(['name'], ['path'])
>>> up.run(None, {"name": "a"}, host={"path": "/tmp/upload"})
{'path': '/tmp/upload', 'name': 'a'}
"""
name: str
fn: Callable[..., dict]
title: str
description: str = ""
effect: str = "write"
runs: str = "now"
host_params: tuple[str, ...] = ()
max_upload_bytes: Optional[int] = None
spends: bool = False
def __post_init__(self):
if not isinstance(self.name, str) or not _OP_NAME_PATTERN.match(self.name):
raise ValueError(
f"genre op name {self.name!r} must be snake_case "
"(lowercase letters, digits, underscores; starting with a letter)"
)
if not callable(self.fn):
raise TypeError(f"genre op {self.name!r}: fn must be callable")
if not isinstance(self.title, str) or not self.title.strip():
raise ValueError(
f"genre op {self.name!r}: title must be a non-empty string"
)
if self.effect not in GENRE_OP_EFFECTS:
raise ValueError(
f"genre op {self.name!r}: effect {self.effect!r} not in "
f"{GENRE_OP_EFFECTS}"
)
if self.runs not in GENRE_OP_RUNS:
raise ValueError(
f"genre op {self.name!r}: runs {self.runs!r} not in {GENRE_OP_RUNS}"
)
if not self.description:
object.__setattr__(self, "description", inspect.getdoc(self.fn) or "")
object.__setattr__(self, "host_params", tuple(self.host_params))
if not isinstance(self.spends, bool):
raise TypeError(
f"genre op {self.name!r}: spends must be a bool, got {self.spends!r}"
)
if self.max_upload_bytes is not None and (
not isinstance(self.max_upload_bytes, int)
or isinstance(self.max_upload_bytes, bool)
or self.max_upload_bytes <= 0
):
raise ValueError(
f"genre op {self.name!r}: max_upload_bytes must be a positive int "
f"or None, got {self.max_upload_bytes!r}"
)
object.__setattr__(
self,
"_model",
_op_params_model(self.fn, name=self.name, host_params=self.host_params),
)
@property
def params_model(self):
"""The pydantic model of the op's parameters (``extra="forbid"``)."""
return self._model
@property
def params_schema(self) -> dict:
"""JSON Schema (an object, ``additionalProperties: false``) of the parameters."""
return self._model.model_json_schema()
[docs]
def validate_params(self, params: Optional[Mapping] = None) -> dict:
"""``params`` checked and coerced against :attr:`params_schema`.
Raises :class:`pydantic.ValidationError` (a :class:`ValueError`) on a missing,
unknown or wrongly typed parameter. Only the parameters the caller passed are
returned, so the op's own defaults stay the op's.
"""
validated = self._model.model_validate(dict(params or {}))
return validated.model_dump(exclude_unset=True)
[docs]
def run(
self,
project,
params: Optional[Mapping] = None,
*,
host: Optional[Mapping] = None,
) -> dict:
"""Run the op on ``project``: ``params`` (the CLIENT's, validated against
:attr:`params_schema`) plus ``host`` (the host's, passed through as given).
``host`` may carry only the op's declared :attr:`host_params`; anything else is
a host bug and raises :class:`TypeError`, as does a required host parameter the
host did not supply. A client parameter that is missing, unknown (a host
parameter included) or mistyped raises :class:`pydantic.ValidationError`.
"""
client = self.validate_params(
params
) # the client's mistakes are reported first
host = dict(host or {})
undeclared = sorted(set(host) - set(self.host_params))
if undeclared:
raise TypeError(
f"genre op {self.name!r} takes no host parameter(s) {undeclared}; "
f"its host parameters are {list(self.host_params)}"
)
missing = [h for h in self._required_host_params() if h not in host]
if missing:
raise TypeError(
f"genre op {self.name!r} needs host parameter(s) {missing} "
"(the host supplies these — an upload's path, say)"
)
return self.fn(project, **client, **host)
def _required_host_params(self) -> list:
sig = inspect.signature(self.fn)
return [
h
for h in self.host_params
if sig.parameters[h].default is inspect.Parameter.empty
]
[docs]
def to_dict(self) -> dict:
"""The op's JSON row: everything but the function."""
return {
"name": self.name,
"title": self.title,
"description": self.description,
"effect": self.effect,
"runs": self.runs,
"params_schema": self.params_schema,
"host_params": list(self.host_params),
"max_upload_bytes": self.max_upload_bytes,
"spends": self.spends,
}
#: Registry of per-genre operations, keyed by genre slug (the owning app registers).
genre_ops_registry: Registry = Registry(name="nw.genre_ops", on_conflict="error")
[docs]
def register_genre_ops(genre_slug: str, ops) -> tuple:
"""Register the operations a genre offers on its projects; returns them as a tuple.
Called once by the genre's **owning app**, beside its project factory, so a host can
serve the genre's project operations via :func:`genre_ops` without importing the
genre's package. Names must be unique within the genre. Registering the same genre
twice raises (the registry refuses conflicts, as every nw genre registry does).
>>> def _peek(project) -> dict:
... '''Say hello.'''
... return {"hello": True}
>>> _ = register_genre_ops("_ops_demo", [GenreOp("peek", _peek, title="Peek",
... effect="read")])
>>> [op.name for op in genre_ops("_ops_demo")]
['peek']
>>> genre_op("_ops_demo", "peek").run(None)
{'hello': True}
>>> genre_ops("_nobody")
()
>>> del genre_ops_registry["_ops_demo"]
"""
_validate_slug(genre_slug, what="genre ops")
ops = tuple(ops)
for op in ops:
if not isinstance(op, GenreOp):
raise TypeError(
f"register_genre_ops expects GenreOp rows, got {type(op).__name__}"
)
names = [op.name for op in ops]
duplicates = sorted({n for n in names if names.count(n) > 1})
if duplicates:
raise ValueError(f"genre {genre_slug!r}: duplicate op names {duplicates}")
genre_ops_registry.register(genre_slug, ops)
return ops
[docs]
def genre_ops(genre_slug: str) -> tuple:
"""The ops registered for ``genre_slug``, in registration order (``()`` if none)."""
return genre_ops_registry[genre_slug] if genre_slug in genre_ops_registry else ()
[docs]
def genre_op(genre_slug: str, name: str) -> GenreOp:
"""The op ``name`` of ``genre_slug``; :class:`UnknownGenreOpError` naming the known."""
for op in genre_ops(genre_slug):
if op.name == name:
return op
known = [op.name for op in genre_ops(genre_slug)]
raise UnknownGenreOpError(
f"genre {genre_slug!r} has no op {name!r}; known ops: {known}"
)
[docs]
def genre_ops_catalogue(genre_slug: str) -> list:
"""The pure-JSON catalogue of ``genre_slug``'s ops (``[]`` for a genre with none).
One dict per op — ``name``, ``title``, ``description``, ``effect``, ``runs``,
``params_schema``, ``host_params``, ``max_upload_bytes``, ``spends`` — the shape a
host exports to a frontend's codegen or an MCP
tool builder.
"""
return [op.to_dict() for op in genre_ops(genre_slug)]
__all__ = [
"GENRE_STATUSES",
"Genre",
"Template",
"genres",
"get_genre",
"list_genres",
"register_genre",
"genre_catalog",
"describe_genre",
"recommend_genre",
"resolve_defaults",
"GenreResolver",
"genre_resolvers",
"register_genre_resolver",
"resolve_genre",
"GenreInitializer",
"genre_initializers",
"register_genre_initializer",
"initialize_genre",
"GenreProjectFactory",
"genre_project_factories",
"register_genre_project_factory",
"has_genre_project_factory",
"can_place_genre_project",
"create_genre_project",
"PLACEMENT_ARG",
"GENRE_OP_EFFECTS",
"GENRE_OP_RUNS",
"GenreOp",
"UnknownGenreOpError",
"GenreOpRefused",
"GenreOpCancelled",
"CANCEL_PARAM",
"genre_ops_registry",
"register_genre_ops",
"genre_ops",
"genre_op",
"genre_ops_catalogue",
]