Python core - `fxq`¶

Reference for the backend core, generated from docstrings and type hints. This is the implementation behind the HTTP API; most integrators should prefer the SDK or the HTTP API rather than importing fxq directly.

Application¶

app ¶

FastAPI application factory.

create_app is the single entry point used by the dev server and the test suite. It wires together the database engine, session factory, media storage, all routers, and optional static SPA serving.

Functions:

Name	Description
`create_app`	Construct and return the fxquinox FastAPI application.

Classes¶

Functions:¶

create_app ¶

create_app(database_url: str | None = None, seed: bool = False, secret: str | None = None, cors_origins: list[str] | None = None, media_root: str | None = None, static_dir: str | None = None) -> FastAPI

Construct and return the fxquinox FastAPI application.

Creates the SQLAlchemy engine, applies the built-in schema seed (when seed=True), registers all routers, and optionally mounts a built SPA from static_dir (or the FXQ_STATIC_DIR env var).

The /healthz endpoint is always available without auth.

Parameters:

Name	Type	Description	Default
`database_url`	`str \| None`	SQLAlchemy connection string. Falls back to `settings.database_url` (SQLite dev DB by default).	`None`
`seed`	`bool`	If `True`, run `seed_builtin_schema` on startup.	`False`
`secret`	`str \| None`	HMAC-SHA-256 secret for JWT signing.	`None`
`cors_origins`	`list[str] \| None`	List of allowed CORS origins. Defaults to the local Vite dev server (`localhost:5173`).	`None`
`media_root`	`str \| None`	Path for local media storage. Falls back to `settings.media_root`.	`None`
`static_dir`	`str \| None`	Directory containing the built SPA (optional).	`None`

Returns:

Type	Description
`FastAPI`	A configured `FastAPI` application instance.

Examples:

>>> app = create_app(database_url="sqlite+pysqlite:///:memory:", seed=True)
>>> app.title
'fxquinox'

Configuration¶

config ¶

Application configuration loaded from environment variables or a .env file.

All variables are prefixed with FXQ_ (e.g. FXQ_DATABASE_URL). Defaults are suitable for local development with SQLite; override for production.

Classes:

Name	Description
`Settings`	Pydantic settings model for fxquinox.

Functions:

Name	Description
`get_settings`	Return a fresh `Settings` instance resolved from the environment.

Classes¶

Settings ¶

Bases: BaseSettings

Pydantic settings model for fxquinox.

Reads FXQ_* environment variables and a .env file (if present). Unknown variables are silently ignored (extra="ignore").

Attributes:

Name	Type	Description
`database_url`	`str`	SQLAlchemy connection string. Defaults to a local SQLite file (`fxq_dev.db`).
`redis_url`	`str`	Redis connection string (reserved for future use).
`s3_endpoint_url`	`str \| None`	Custom S3-compatible endpoint, or `None` for AWS.
`s3_bucket`	`str`	Bucket name for object storage.
`media_root`	`str`	Filesystem path for the local media store.
`debug`	`bool`	Enable verbose debug output.

Examples:

>>> s = Settings(database_url="sqlite+pysqlite:///:memory:")
>>> s.database_url
'sqlite+pysqlite:///:memory:'

Attributes¶

model_config `class-attribute` `instance-attribute` ¶

model_config = SettingsConfigDict(env_prefix='FXQ_', env_file='.env', extra='ignore')

database_url `class-attribute` `instance-attribute` ¶

database_url: str = 'sqlite+pysqlite:///./fxq_dev.db'

redis_url `class-attribute` `instance-attribute` ¶

redis_url: str = 'redis://localhost:6379/0'

s3_endpoint_url `class-attribute` `instance-attribute` ¶

s3_endpoint_url: str | None = None

s3_bucket `class-attribute` `instance-attribute` ¶

s3_bucket: str = 'fxquinox'

media_root `class-attribute` `instance-attribute` ¶

media_root: str = './media_store'

secret `class-attribute` `instance-attribute` ¶

secret: str = ''

cors_origins `class-attribute` `instance-attribute` ¶

cors_origins: str = ''

debug `class-attribute` `instance-attribute` ¶

debug: bool = False

Functions:¶

get_settings ¶

get_settings() -> Settings

Return a fresh Settings instance resolved from the environment.

Called at startup and in tests; cheap to construct since pydantic-settings caches nothing — callers that need a singleton should cache the result.

Returns:

Type	Description
`Settings`	A fully-resolved `Settings` object.

Examples:

>>> s = get_settings()
>>> isinstance(s, Settings)
True

Database¶

db ¶

SQLAlchemy engine and session-factory helpers.

Base is the shared declarative base inherited by every ORM model. make_engine and make_session_factory are called once at application startup by create_app; tests create their own in-memory instances.

Classes:

Name	Description
`Base`	Shared SQLAlchemy declarative base for all fxquinox ORM models.

Functions:

Name	Description
`make_engine`	Create a SQLAlchemy engine with sensible defaults for the dialect.
`make_session_factory`	Create a `sessionmaker` bound to the given engine.

Classes¶

Base ¶

Bases: DeclarativeBase

Shared SQLAlchemy declarative base for all fxquinox ORM models.

Functions:¶

make_engine ¶

make_engine(database_url: str) -> Engine

Create a SQLAlchemy engine with sensible defaults for the dialect.

SQLite gets check_same_thread=False so FastAPI's threadpool workers can share the connection. In-memory SQLite additionally uses StaticPool so all threads see the same database (needed for tests and the dev server when database_url is ":memory:").

Parameters:

Name	Type	Description	Default
`database_url`	`str`	A SQLAlchemy-style connection string, e.g. `"sqlite+pysqlite:///./fxq_dev.db"` or a Postgres DSN.	required

Returns:

Type	Description
`Engine`	A configured `Engine` instance.

Examples:

>>> from fxq.db import make_engine
>>> e = make_engine("sqlite+pysqlite:///:memory:")
>>> e.url.get_backend_name()
'sqlite'

make_session_factory ¶

make_session_factory(engine: Engine) -> sessionmaker

Create a sessionmaker bound to the given engine.

autoflush=False and expire_on_commit=False are deliberate: the API layer controls flush/commit explicitly, and we do not want SQLAlchemy to lazily re-fetch objects after a commit (they are serialised to JSON immediately).

Parameters:

Name	Type	Description	Default
`engine`	`Engine`	A connected SQLAlchemy `Engine`.	required

Returns:

Type	Description
`sessionmaker`	A `sessionmaker` that produces `Session` objects on call.

Examples:

>>> from fxq.db import make_engine, make_session_factory
>>> engine = make_engine("sqlite+pysqlite:///:memory:")
>>> factory = make_session_factory(engine)
>>> with factory() as s:
...     s.execute(__import__('sqlalchemy').text('SELECT 1'))
<...>

Schema engine¶

types ¶

Immutable in-memory schema representation built from the database.

These frozen dataclasses are populated once per request by schema.loader.load_schema and passed to the query compiler. They are read-only views — mutations go through services.schema_admin.

Classes:

Name	Description
`FieldDefn`	Immutable snapshot of one field's definition.
`EntityDefn`	Immutable snapshot of one entity type's definition.
`Schema`	Immutable snapshot of all entity type definitions for one request.

Attributes:

Name	Type	Description
`LINK_TYPES`

Attributes¶

LINK_TYPES `module-attribute` ¶

LINK_TYPES = {'entity_link', 'multi_entity_link'}

Classes¶

FieldDefn `dataclass` ¶

FieldDefn(code: str, display_name: str, data_type: str, config: dict = dict(), is_required: bool = False, inherit: bool = False)

Immutable snapshot of one field's definition.

Attributes:

Name	Type	Description
`code`	`str`	Field identifier (e.g. `"status"`, `"step"`).
`display_name`	`str`	Human-readable label.
`data_type`	`str`	Storage/widget type (e.g. `"text"`, `"enum"`, `"entity_link"`).
`config`	`dict`	Type-specific configuration dict (link types, enum options, …).
`is_required`	`bool`	Whether the field must have a value.
`inherit`	`bool`	Whether the value is inherited from the parent entity.

Attributes¶

code `instance-attribute` ¶

code: str

display_name `instance-attribute` ¶

display_name: str

data_type `instance-attribute` ¶

data_type: str

config `class-attribute` `instance-attribute` ¶

config: dict = field(default_factory=dict)

is_required `class-attribute` `instance-attribute` ¶

is_required: bool = False

inherit `class-attribute` `instance-attribute` ¶

inherit: bool = False

is_link `property` ¶

is_link: bool

True if this field holds a link to another entity.

Examples:

>>> FieldDefn("entity", "Entity", "entity_link").is_link
True
>>> FieldDefn("name", "Name", "text").is_link
False

link_types `property` ¶

link_types: list[str]

The entity types this link may point to (from config["link_types"]).

Returns an empty list for non-link fields.

Examples:

>>> FieldDefn("task", "Task", "entity_link", {"link_types": ["Task"]}).link_types
['Task']
>>> FieldDefn("name", "Name", "text").link_types
[]

EntityDefn `dataclass` ¶

EntityDefn(code: str, display_name: str, hierarchy_mode: str = 'none', parent_field: str | None = None, fields: dict[str, FieldDefn] = field(default_factory=dict))

Immutable snapshot of one entity type's definition.

Attributes:

Name	Type	Description
`code`	`str`	Unique entity code (e.g. `"Task"`).
`display_name`	`str`	Human-readable name.
`hierarchy_mode`	`str`	`"none"`, `"flat"`, or `"nested"`.
`parent_field`	`str \| None`	Field code pointing to the parent entity, or `None`.
`fields`	`dict[str, FieldDefn]`	Dict of field code → `FieldDefn`.

Methods:

Name	Description
`field`	Return the named field definition.
`has_field`	Return `True` if this entity type has a field with the given code.

Attributes¶

code `instance-attribute` ¶

code: str

display_name `instance-attribute` ¶

display_name: str

hierarchy_mode `class-attribute` `instance-attribute` ¶

hierarchy_mode: str = 'none'

parent_field `class-attribute` `instance-attribute` ¶

parent_field: str | None = None

fields `class-attribute` `instance-attribute` ¶

fields: dict[str, FieldDefn] = field(default_factory=dict)

Methods:¶

field ¶

field(code: str) -> FieldDefn

Return the named field definition.

Parameters:

Name	Type	Description	Default
`code`	`str`	Field identifier.	required

Returns:

Type	Description
`FieldDefn`	The corresponding `FieldDefn`.

Raises:

Type	Description
`KeyError`	If no field with that code exists on this entity type.

Examples:

>>> et = EntityDefn("Task", "Task", fields={"name": FieldDefn("name", "Name", "text")})
>>> et.field("name").data_type
'text'

has_field ¶

has_field(code: str) -> bool

Return True if this entity type has a field with the given code.

Examples:

>>> et = EntityDefn("Task", "Task", fields={"name": FieldDefn("name", "Name", "text")})
>>> et.has_field("name")
True
>>> et.has_field("missing")
False

Schema `dataclass` ¶

Schema(entities: dict[str, EntityDefn] = dict())

Immutable snapshot of all entity type definitions for one request.

Built by schema.loader.load_schema from the database; passed to the query compiler and any service that needs to inspect field types.

Attributes:

Name	Type	Description
`entities`	`dict[str, EntityDefn]`	Dict of entity code → `EntityDefn`.

Methods:

Name	Description
`entity`	Return the definition for the named entity type.
`has_entity`	Return `True` if the schema contains the named entity type.

Attributes¶

entities `class-attribute` `instance-attribute` ¶

entities: dict[str, EntityDefn] = field(default_factory=dict)

Methods:¶

entity ¶

entity(code: str) -> EntityDefn

Return the definition for the named entity type.

Parameters:

Name	Type	Description	Default
`code`	`str`	Entity type code (e.g. `"Shot"`).	required

Returns:

Type	Description
`EntityDefn`	The corresponding `EntityDefn`.

Raises:

Type	Description
`KeyError`	If the entity type is not in the schema.

Examples:

>>> s = Schema({"Task": EntityDefn("Task", "Task")})
>>> s.entity("Task").code
'Task'

has_entity ¶

has_entity(code: str) -> bool

Return True if the schema contains the named entity type.

Examples:

>>> s = Schema({"Task": EntityDefn("Task", "Task")})
>>> s.has_entity("Task")
True
>>> s.has_entity("Nope")
False

loader ¶

Load the current schema from the database into immutable dataclasses.

load_schema is called on every query request so the in-memory view is always up to date with any runtime schema changes. The result is a lightweight, frozen Schema object safe to pass across the call stack.

Functions:

Name	Description
`load_schema`	Build a `Schema` snapshot from all `EntityType` rows in the DB.

Classes¶

Functions:¶

load_schema ¶

load_schema(session: Session) -> Schema

Build a Schema snapshot from all EntityType rows in the DB.

Reads every EntityType and its associated FieldDef rows (via the eagerly loaded relationship) and converts them to frozen dataclasses.

Parameters:

Name	Type	Description	Default
`session`	`Session`	An open SQLAlchemy session.	required

Returns:

Type	Description
`Schema`	A `Schema` containing an `EntityDefn` for every registered type.

Examples:

>>> schema = load_schema(session)
>>> schema.has_entity("Task")
True

Query DSL¶

dsl ¶

Filter DSL — a JSON-serialisable query language for the entity store.

Filters are nested Python lists that mirror a simple expression tree:

A leaf condition: [path, op, value] e.g. ["status", "is", "inProgress"]
A logical group: ["and", child1, child2, ...] or ["or", ...]

The path string supports dot-notation traversal through link fields: "entity.Shot.sequence" joins the linked Shot's sequence.

parse_filters converts one of these lists into a Condition or Group dataclass tree that the compiler can turn into a SQLAlchemy clause.

Classes:

Name	Description
`DslError`	Raised when a filter list is structurally invalid.
`Condition`	A single leaf predicate.
`Group`	A logical conjunction or disjunction of child nodes.

Functions:

Name	Description
`parse_filters`	Parse a filter list into a `Condition` or `Group` tree.

Attributes:

Name	Type	Description
`OPERATORS`

Attributes¶

OPERATORS `module-attribute` ¶

OPERATORS = {'is', 'is_not', 'in', 'not_in', 'contains', 'starts_with', 'ends_with', 'gte', 'lte', 'gt', 'lt', 'between', 'is_null'}

Classes¶

DslError ¶

Bases: ValueError

Raised when a filter list is structurally invalid.

Examples:

>>> raise DslError("bad filter")

Condition `dataclass` ¶

Condition(path: list[str], op: str, value: object)

A single leaf predicate.

Attributes:

Name	Type	Description
`path`	`list[str]`	The field path split on `"."`, e.g. `["status"]` or `["entity", "Shot", "sequence"]` for a cross-link path.
`op`	`str`	One of the strings in `OPERATORS`.
`value`	`object`	The right-hand side of the comparison. `None` for zero-value operators like `is_null`.

Attributes¶

path `instance-attribute` ¶

path: list[str]

op `instance-attribute` ¶

op: str

value `instance-attribute` ¶

value: object

Group `dataclass` ¶

Group(op: str, children: list[Condition | Group])

A logical conjunction or disjunction of child nodes.

Attributes:

Name	Type	Description
`op`	`str`	`"and"` or `"or"`.
`children`	`list[Condition \| Group]`	One or more `Condition` or nested `Group` nodes.

Attributes¶

op `instance-attribute` ¶

op: str

children `instance-attribute` ¶

children: list[Condition | Group]

Functions:¶

parse_filters ¶

parse_filters(node: list) -> Condition | Group

Parse a filter list into a Condition or Group tree.

Parameters:

Name	Type	Description	Default
`node`	`list`	A non-empty list in DSL form. Either a logical group `["and"\|"or", child, ...]` or a leaf condition `[path, op]` / `[path, op, value]`.	required

Returns:

Type	Description
`Condition \| Group`	A `Group` for logical operators, a `Condition` for leaf nodes.

Raises:

Type	Description
`DslError`	If the structure is invalid (wrong type, unknown operator, empty group, etc.).

Examples:

>>> parse_filters(["status", "is", "inProgress"])
Condition(path=['status'], op='is', value='inProgress')
>>> parse_filters(["and", ["status", "is", "inProgress"], ["step", "is", "anim"]])
Group(op='and', children=[Condition(path=['status'], op='is', value='inProgress'), Condition(path=['step'], op='is', value='anim')])
>>> parse_filters(["status", "is_null"])
Condition(path=['status'], op='is_null', value=None)

compiler ¶

SQL compiler for the filter DSL.

Translates a Condition / Group tree produced by query.dsl.parse_filters into a SQLAlchemy select statement against the Record table. Link-field traversal (entity.Shot.sequence) is handled by joining aliased Record rows at compile time.

Functions:

Name	Description
`build_select`	Build a SQLAlchemy `select` for active records of one entity type.
`run_query`	Execute a DSL query and return matching `Record` objects.

Classes¶

Functions:¶

build_select ¶

build_select(schema: Schema, entity_code: str, filters: Condition | Group | None = None, sort: list[tuple[str, str]] | None = None, limit: int | None = None, offset: int | None = None)

Build a SQLAlchemy select for active records of one entity type.

Applies optional filters (via the DSL compiler), sort columns, and pagination. The base WHERE always requires active = True and entity_type = entity_code.

Parameters:

Name	Type	Description	Default
`schema`	`Schema`	The loaded schema used to resolve field data types.	required
`entity_code`	`str`	The entity type to query (e.g. `"Task"`).	required
`filters`	`Condition \| Group \| None`	Optional parsed filter tree from `parse_filters`.	`None`
`sort`	`list[tuple[str, str]] \| None`	List of `(field_code, direction)` tuples, where `direction` is `"asc"` or `"desc"`.	`None`
`limit`	`int \| None`	Maximum number of rows to return.	`None`
`offset`	`int \| None`	Number of rows to skip (for pagination).	`None`

Returns:

Type	Description
	A SQLAlchemy `Select` statement ready for execution.

Examples:

>>> build_select(schema, "Task", limit=10)

run_query ¶

run_query(session: Session, schema: Schema, entity_code: str, filters: Condition | Group | None = None, sort: list[tuple[str, str]] | None = None, limit: int | None = None, offset: int | None = None) -> list[Record]

Execute a DSL query and return matching Record objects.

Convenience wrapper that calls build_select and executes it against the given session.

Parameters:

Name	Type	Description	Default
`session`	`Session`	An open SQLAlchemy session.	required
`schema`	`Schema`	The loaded schema.	required
`entity_code`	`str`	Entity type to query.	required
`filters`	`Condition \| Group \| None`	Optional parsed filter tree.	`None`
`sort`	`list[tuple[str, str]] \| None`	Optional sort specification.	`None`
`limit`	`int \| None`	Optional row limit.	`None`
`offset`	`int \| None`	Optional row offset.	`None`

Returns:

Type	Description
`list[Record]`	A list of `Record` instances matching the query.

Examples:

>>> run_query(session, schema, "Task", limit=5)

Publish chain¶

service ¶

Publish service: the Task → Product → Version → Representation chain.

A publish creates (or reuses) a Product for the named product on a task, then appends a new Version with an auto-incremented version number, and creates one Representation + FileLocation per supplied spec.

Functions:

Name	Description
`publish`	Publish a new version of a product on a task.

Classes¶

Functions:¶

publish ¶

publish(session: Session, *, task_id: str, product: str, product_type: str, representations: list[dict], author: str | None = None, comment: str | None = None, inputs: list[str] | None = None, source_workfile_id: str | None = None, frame_start: int | None = None, frame_end: int | None = None, fps: float | None = None, site_id: str | None = None) -> dict

Publish a new version of a product on a task.

Finds or creates the Product, increments the version number, creates a Version record with status="waitingToStart", and creates a Representation + FileLocation for each supplied spec.

Parameters:

Name	Type	Description	Default
`session`	`Session`	Open DB session.	required
`task_id`	`str`	UUID of the parent `Task`.	required
`product`	`str`	Product name (e.g. `"compMain"`); created if absent.	required
`product_type`	`str`	camelCase product-type code (e.g. `"compositing"`).	required
`representations`	`list[dict]`	List of representation specs, each a dict with `name`, optional `files`, `for_review`, and `tags`.	required
`author`	`str \| None`	Display name of the publishing artist.	`None`
`comment`	`str \| None`	Optional publish comment.	`None`
`inputs`	`list[str] \| None`	Optional list of input `Version` UUIDs (for lineage).	`None`
`source_workfile_id`	`str \| None`	Optional UUID of the source `Workfile`.	`None`
`frame_start`	`int \| None`	First frame number of the published range.	`None`
`frame_end`	`int \| None`	Last frame number.	`None`
`fps`	`float \| None`	Frames per second.	`None`
`site_id`	`str \| None`	Override the site for `FileLocation` records.	`None`

Returns:

Type	Description
`dict`	A dict with `product`, `version`, and `representations` keys,
`dict`	each containing `id` and `data`.

Examples:

>>> result = publish(session, task_id=task_id, product="animMain", product_type="animation", representations=[{"name": "exr", "files": []}])
>>> result["version"]["data"]["number"]
1

lineage ¶

Version lineage: upstream (inputs) and downstream (dependents) resolution.

Lineage is stored as Version.data["inputs"] — a list of {type, id} link dicts. Upstream walks those links; downstream scans all versions for any that list the target as an input.

Functions:

Name	Description
`upstream`	Return the `Version` records that are direct inputs of `version_id`.
`dependency_tree`	The full upstream dependency tree of a version (VAL-51).
`downstream`	Return all active `Version` records that list `version_id` as an input.

Classes¶

Functions:¶

upstream ¶

upstream(session: Session, version_id: str) -> list[Record]

Return the Version records that are direct inputs of version_id.

Preserves the order declared in data["inputs"].

Parameters:

Name	Type	Description	Default
`session`	`Session`	Open DB session.	required
`version_id`	`str`	UUID of the version to inspect.	required

Returns:

Type	Description
`list[Record]`	List of input `Version` records in input-list order. Returns an
`list[Record]`	empty list if the version does not exist or has no inputs.

Examples:

>>> upstream(session, version_id)
[<Record entity_type='Version' ...>]

dependency_tree ¶

dependency_tree(session: Session, version_id: str, max_depth: int = 12) -> dict | None

The full upstream dependency tree of a version (VAL-51).

Each node carries enough to render a usable inspector — label, product, task, status, author, resolved path — plus its recursive inputs. Cycles are broken with a cycle flag; depth is capped at max_depth.

Returns None if the version does not exist.

downstream ¶

downstream(session: Session, version_id: str) -> list[Record]

Return all active Version records that list version_id as an input.

Parameters:

Name	Type	Description	Default
`session`	`Session`	Open DB session.	required
`version_id`	`str`	UUID of the version to find dependents of.	required

Returns:

Type	Description
`list[Record]`	List of `Version` records that depend on this version. Order is
`list[Record]`	unspecified (table scan order).

Examples:

>>> downstream(session, version_id)
[]

paths ¶

Path template resolution for the publish pipeline.

Templates use Python str.format syntax with a {root} variable automatically filled from the current site's roots dict. Example template::

"{root}/projects/{project_code}/{entity_type}/{entity_code}/{step}/v{version:03d}"

Functions:

Name	Description
`current_platform`	Return the canonical platform name for the current machine.
`resolve`	Resolve a path template with the given context dict.
`resolve_for_site`	Resolve a path template using the site root for the current platform.

Functions:¶

current_platform ¶

current_platform() -> str

Return the canonical platform name for the current machine.

Returns:

Type	Description
`str`	One of `"windows"`, `"linux"`, or `"darwin"`. Unknown
`str`	platforms fall back to `"linux"`.

Examples:

>>> current_platform() in ("windows", "linux", "darwin")
True

resolve ¶

resolve(template: str, context: dict) -> str

Resolve a path template with the given context dict.

Parameters:

Name	Type	Description	Default
`template`	`str`	A `str.format`-style template.	required
`context`	`dict`	Keyword arguments for template substitution.	required

Returns:

Type	Description
`str`	The resolved path string.

Raises:

Type	Description
`KeyError`	If the template references a variable not in `context`.

Examples:

>>> resolve("{root}/projects/{code}", {"root": "/mnt/fxq", "code": "demo"})
'/mnt/fxq/projects/demo'

resolve_for_site ¶

resolve_for_site(template: str, context: dict, site_roots: dict, platform: str | None = None) -> str

Resolve a path template using the site root for the current platform.

Looks up the platform root in site_roots (falling back to the "linux" entry) and injects it as {root} before resolving.

Parameters:

Name	Type	Description	Default
`template`	`str`	A `str.format`-style template containing `{root}`.	required
`context`	`dict`	Additional template variables.	required
`site_roots`	`dict`	Dict of `platform → root_path` (e.g. `{"windows": "C:/fxq/projects", "linux": "/mnt/fxq/projects"}`).	required
`platform`	`str \| None`	Override for the auto-detected platform.	`None`

Returns:

Type	Description
`str`	The resolved filesystem path string.

Examples:

>>> resolve_for_site("{root}/{code}", {"code": "demo"}, {"linux": "/mnt/fxq"}, "linux")
'/mnt/fxq/demo'
>>> resolve_for_site("{root}/{code}", {"code": "demo"}, {"windows": "C:/fxq"}, "windows")
'C:/fxq/demo'

Models¶

record ¶

The universal Record table — the single-table store for every entity.

Every fxquinox entity (Project, Shot, Asset, Task, User, Note, …) is a row in this one table. Entity-specific fields live in the data JSON column so new types require no schema migration.

Classes:

Name	Description
`Record`	Universal entity row.

Classes¶

Record ¶

Bases: Base

Universal entity row.

Every fxquinox entity maps to one Record. Structural metadata (entity_type, project_id, status, active) is stored in real columns for efficient filtering; all other entity-specific fields live in the data JSON blob.

Attributes:

Name	Type	Description
`id`	`Mapped[str]`	UUID primary key.
`entity_type`	`Mapped[str]`	Canonical entity code, e.g. `"Task"`, `"Shot"`.
`project_id`	`Mapped[str \| None]`	Foreign-key-style link to a `Project` record (`None` for project-level and global entities).
`status`	`Mapped[str \| None]`	Current status code, e.g. `"inProgress"`, `"final"`.
`data`	`Mapped[dict]`	JSON dict holding all entity-specific fields.
`active`	`Mapped[bool]`	`False` means soft-deleted.
`thumbnail_id`	`Mapped[str \| None]`	Key of the thumbnail in the media store.
`created_by`	`Mapped[str \| None]`	ID of the user (or API key) that created this record.
`created_at`	`Mapped[datetime]`	UTC creation timestamp.
`updated_at`	`Mapped[datetime]`	UTC last-modified timestamp, updated automatically.

Attributes¶

id `class-attribute` `instance-attribute` ¶

id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid)

entity_type `class-attribute` `instance-attribute` ¶

entity_type: Mapped[str] = mapped_column(String(64), index=True)

project_id `class-attribute` `instance-attribute` ¶

project_id: Mapped[str | None] = mapped_column(String(36), index=True)

status `class-attribute` `instance-attribute` ¶

status: Mapped[str | None] = mapped_column(String(64), index=True)

data `class-attribute` `instance-attribute` ¶

data: Mapped[dict] = mapped_column(JSON, default=dict)

active `class-attribute` `instance-attribute` ¶

active: Mapped[bool] = mapped_column(Boolean, default=True)

thumbnail_id `class-attribute` `instance-attribute` ¶

thumbnail_id: Mapped[str | None] = mapped_column(String(36))

created_by `class-attribute` `instance-attribute` ¶

created_by: Mapped[str | None] = mapped_column(String(36))

created_at `class-attribute` `instance-attribute` ¶

created_at: Mapped[datetime] = mapped_column(DateTime, default=_now)

updated_at `class-attribute` `instance-attribute` ¶

updated_at: Mapped[datetime] = mapped_column(DateTime, default=_now, onupdate=_now)

meta ¶

Schema-metadata ORM models (entity types, field definitions, statuses, etc.).

These tables define the shape of the data in the record table. They are populated at startup by schema.seed.seed_builtin_schema and can be extended at runtime through services.schema_admin.

Classes:

Name	Description
`EntityType`	A registered entity type (e.g. `Shot`, `Asset`, `Task`).
`FieldDef`	A field definition for one field on an `EntityType`.
`Status`	A configurable status option (e.g. `waitingToStart`, `inProgress`).
`Workflow`	A global status transition edge for one entity type.
`Step`	A configurable pipeline/department step (Layout, Anim, Comp, ...).
`Priority`	A configurable task priority level (None, Low, Medium, High, Urgent).
`Setting`	Singleton-ish key/value app settings (e.g. custom branding).

Classes¶

EntityType ¶

Bases: Base

A registered entity type (e.g. Shot, Asset, Task).

Built-in types are seeded once and cannot be deleted. Custom types created via the schema-admin API have is_builtin=False and may be removed when no records reference them.

Attributes:

Name	Type	Description
`id`	`Mapped[str]`	UUID primary key.
`code`	`Mapped[str]`	Unique identifier used in the API and stored in `Record.entity_type` (e.g. `"Shot"`).
`display_name`	`Mapped[str]`	Human-readable label.
`is_builtin`	`Mapped[bool]`	`True` for types shipped with fxquinox.
`icon`	`Mapped[str \| None]`	Material Symbol icon key rendered in the sidebar / entity picker.
`hierarchy_mode`	`Mapped[str]`	`"none"` (flat list), `"flat"` (one level of grouping), or `"nested"` (full tree via `parent_field`).
`parent_field`	`Mapped[str \| None]`	Name of the `entity_link` field that points to the parent (e.g. `"sequence"` for Shot, `"episode"` for Sequence).
`fields`	`Mapped[list[FieldDef]]`	Ordered list of `FieldDef` rows for this entity type.

Attributes¶

id `class-attribute` `instance-attribute` ¶

id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid)

code `class-attribute` `instance-attribute` ¶

code: Mapped[str] = mapped_column(String(64), unique=True, index=True)

display_name `class-attribute` `instance-attribute` ¶

display_name: Mapped[str] = mapped_column(String(128))

is_builtin `class-attribute` `instance-attribute` ¶

is_builtin: Mapped[bool] = mapped_column(Boolean, default=False)

icon `class-attribute` `instance-attribute` ¶

icon: Mapped[str | None] = mapped_column(String(64))

hierarchy_mode `class-attribute` `instance-attribute` ¶

hierarchy_mode: Mapped[str] = mapped_column(String(16), default='none')

parent_field `class-attribute` `instance-attribute` ¶

parent_field: Mapped[str | None] = mapped_column(String(64))

fields `class-attribute` `instance-attribute` ¶

fields: Mapped[list[FieldDef]] = relationship(back_populates='entity_type', cascade='all, delete-orphan', order_by='FieldDef.sort_order, FieldDef.code')

FieldDef ¶

Bases: Base

A field definition for one field on an EntityType.

The data_type drives how the value is stored in Record.data and how the frontend renders the editor widget. config is a free-form dict whose shape depends on data_type (e.g. {"link_types": [...]}, {"options": [...], "option_labels": {...}} for enums).

Attributes:

Name	Type	Description
`id`	`Mapped[str]`	UUID primary key.
`entity_type_id`	`Mapped[str]`	FK to `EntityType.id`.
`code`	`Mapped[str]`	Field identifier, unique within the entity type.
`display_name`	`Mapped[str]`	Human-readable column label.
`data_type`	`Mapped[str]`	One of the types in `schema_admin.VALID_DATA_TYPES`.
`config`	`Mapped[dict]`	Type-specific configuration dict.
`is_builtin`	`Mapped[bool]`	`True` for fields shipped with fxquinox.
`is_required`	`Mapped[bool]`	Whether the field must have a value.
`sort_order`	`Mapped[int]`	Determines column order in the UI (definition order for built-ins; appended for custom fields).
`default`	`Mapped[dict \| None]`	Optional default value stored as a JSON scalar/object.
`inherit`	`Mapped[bool]`	Whether the field value is inherited from the parent entity.
`entity_type`	`Mapped[EntityType]`	Back-reference to the owning `EntityType`.

Attributes¶

id `class-attribute` `instance-attribute` ¶

id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid)

entity_type_id `class-attribute` `instance-attribute` ¶

entity_type_id: Mapped[str] = mapped_column(ForeignKey('entity_type.id'), index=True)

code `class-attribute` `instance-attribute` ¶

code: Mapped[str] = mapped_column(String(64))

display_name `class-attribute` `instance-attribute` ¶

display_name: Mapped[str] = mapped_column(String(128))

data_type `class-attribute` `instance-attribute` ¶

data_type: Mapped[str] = mapped_column(String(32))

config `class-attribute` `instance-attribute` ¶

config: Mapped[dict] = mapped_column(JSON, default=dict)

is_builtin `class-attribute` `instance-attribute` ¶

is_builtin: Mapped[bool] = mapped_column(Boolean, default=False)

is_required `class-attribute` `instance-attribute` ¶

is_required: Mapped[bool] = mapped_column(Boolean, default=False)

sort_order `class-attribute` `instance-attribute` ¶

sort_order: Mapped[int] = mapped_column(default=0)

default `class-attribute` `instance-attribute` ¶

default: Mapped[dict | None] = mapped_column(JSON)

inherit `class-attribute` `instance-attribute` ¶

inherit: Mapped[bool] = mapped_column(Boolean, default=False)

entity_type `class-attribute` `instance-attribute` ¶

entity_type: Mapped[EntityType] = relationship(back_populates='fields')

Status ¶

Bases: Base

A configurable status option (e.g. waitingToStart, inProgress).

Status codes are camelCase by convention. The state field maps each code to one of five lifecycle buckets used by the frontend for grouping and colour-coding: not_started, in_progress, blocked, done, cancelled.

Global statuses (entity_type_id=None) apply to all entity types. Entity-scoped statuses (entity_type_id set) are restricted to that type.

Attributes:

Name	Type	Description
`id`	`Mapped[str]`	UUID primary key.
`code`	`Mapped[str]`	Unique camelCase code stored in `Record.status` (e.g. `"pendingReview"`).
`name`	`Mapped[str]`	Human-readable label (e.g. `"Pending Review"`).
`color`	`Mapped[str]`	Hex colour rendered in status badges.
`icon`	`Mapped[str \| None]`	Material Symbol icon key.
`short_name`	`Mapped[str \| None]`	Abbreviated label for compact displays (e.g. `"REV"`).
`state`	`Mapped[str]`	Lifecycle bucket — one of `not_started`, `in_progress`, `blocked`, `done`, `cancelled`.
`is_done`	`Mapped[bool]`	`True` when this status represents completion.
`is_wip`	`Mapped[bool]`	`True` when active work is in progress.
`is_retake`	`Mapped[bool]`	`True` for retake/redo statuses.
`is_feedback_request`	`Mapped[bool]`	`True` when the status requests a review.
`entity_type_id`	`Mapped[str \| None]`	Optional FK scoping the status to one entity type.

Attributes¶

id `class-attribute` `instance-attribute` ¶

id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid)

code `class-attribute` `instance-attribute` ¶

code: Mapped[str] = mapped_column(String(64))

name `class-attribute` `instance-attribute` ¶

name: Mapped[str] = mapped_column(String(128))

color `class-attribute` `instance-attribute` ¶

color: Mapped[str] = mapped_column(String(16), default='#888888')

icon `class-attribute` `instance-attribute` ¶

icon: Mapped[str | None] = mapped_column(String(64))

short_name `class-attribute` `instance-attribute` ¶

short_name: Mapped[str | None] = mapped_column(String(16))

state `class-attribute` `instance-attribute` ¶

state: Mapped[str] = mapped_column(String(16), default='in_progress')

is_done `class-attribute` `instance-attribute` ¶

is_done: Mapped[bool] = mapped_column(Boolean, default=False)

is_wip `class-attribute` `instance-attribute` ¶

is_wip: Mapped[bool] = mapped_column(Boolean, default=False)

is_retake `class-attribute` `instance-attribute` ¶

is_retake: Mapped[bool] = mapped_column(Boolean, default=False)

is_feedback_request `class-attribute` `instance-attribute` ¶

is_feedback_request: Mapped[bool] = mapped_column(Boolean, default=False)

entity_type_id `class-attribute` `instance-attribute` ¶

entity_type_id: Mapped[str | None] = mapped_column(ForeignKey('entity_type.id'))

Workflow ¶

Bases: Base

A global status transition edge for one entity type.

Each row permits moving from from_status to to_status for the entity type identified by entity_type_id. role_gate (currently unused by the enforcement layer) is reserved for future role-restricted transitions.

Per-project overrides are stored in Project.data["overrides"]["transitions"] rather than in this table.

Attributes:

Name	Type	Description
`id`	`Mapped[str]`	UUID primary key.
`entity_type_id`	`Mapped[str]`	FK to the entity type this edge belongs to.
`from_status`	`Mapped[str]`	Source status code (e.g. `"inProgress"`).
`to_status`	`Mapped[str]`	Target status code (e.g. `"pendingReview"`).
`role_gate`	`Mapped[str \| None]`	Optional role required to perform this transition.

Attributes¶

id `class-attribute` `instance-attribute` ¶

id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid)

entity_type_id `class-attribute` `instance-attribute` ¶

entity_type_id: Mapped[str] = mapped_column(ForeignKey('entity_type.id'), index=True)

from_status `class-attribute` `instance-attribute` ¶

from_status: Mapped[str] = mapped_column(String(64))

to_status `class-attribute` `instance-attribute` ¶

to_status: Mapped[str] = mapped_column(String(64))

role_gate `class-attribute` `instance-attribute` ¶

role_gate: Mapped[str | None] = mapped_column(String(64))

Step ¶

Bases: Base

A configurable pipeline/department step (Layout, Anim, Comp, ...).

Global list (not per entity type); referenced by the Task.step field and rendered with the configured color across the UI.

Attributes:

Name	Type	Description
`id`	`Mapped[str]`
`code`	`Mapped[str]`
`name`	`Mapped[str]`
`color`	`Mapped[str]`
`icon`	`Mapped[str \| None]`
`sort_order`	`Mapped[int]`

Attributes¶

id `class-attribute` `instance-attribute` ¶

id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid)

code `class-attribute` `instance-attribute` ¶

code: Mapped[str] = mapped_column(String(64))

name `class-attribute` `instance-attribute` ¶

name: Mapped[str] = mapped_column(String(128))

color `class-attribute` `instance-attribute` ¶

color: Mapped[str] = mapped_column(String(16), default='#888888')

icon `class-attribute` `instance-attribute` ¶

icon: Mapped[str | None] = mapped_column(String(64))

sort_order `class-attribute` `instance-attribute` ¶

sort_order: Mapped[int] = mapped_column(default=0)

Priority ¶

Bases: Base

A configurable task priority level (None, Low, Medium, High, Urgent). Global list with a color and a numeric value for sorting/reporting.

Attributes:

Name	Type	Description
`id`	`Mapped[str]`
`code`	`Mapped[str]`
`name`	`Mapped[str]`
`color`	`Mapped[str]`
`icon`	`Mapped[str \| None]`
`value`	`Mapped[int]`
`sort_order`	`Mapped[int]`

Attributes¶

id `class-attribute` `instance-attribute` ¶

id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid)

code `class-attribute` `instance-attribute` ¶

code: Mapped[str] = mapped_column(String(64))

name `class-attribute` `instance-attribute` ¶

name: Mapped[str] = mapped_column(String(128))

color `class-attribute` `instance-attribute` ¶

color: Mapped[str] = mapped_column(String(16), default='#888888')

icon `class-attribute` `instance-attribute` ¶

icon: Mapped[str | None] = mapped_column(String(64))

value `class-attribute` `instance-attribute` ¶

value: Mapped[int] = mapped_column(default=0)

sort_order `class-attribute` `instance-attribute` ¶

sort_order: Mapped[int] = mapped_column(default=0)

Setting ¶

Bases: Base

Singleton-ish key/value app settings (e.g. custom branding).

Attributes:

Name	Type	Description
`key`	`Mapped[str]`
`value`	`Mapped[dict]`

Attributes¶

key `class-attribute` `instance-attribute` ¶

key: Mapped[str] = mapped_column(String(64), primary_key=True)

value `class-attribute` `instance-attribute` ¶

value: Mapped[dict] = mapped_column(JSON, default=dict)

event ¶

Immutable audit log for every create / update / delete operation.

EventLog rows are written by services.records alongside every mutation and are also fanned-out via the in-process event bus to any live WebSocket subscribers.

Classes:

Name	Description
`EventLog`	A single immutable audit-log entry for one entity mutation.

Classes¶

EventLog ¶

Bases: Base

A single immutable audit-log entry for one entity mutation.

Attributes:

Name	Type	Description
`id`	`Mapped[str]`	UUID primary key.
`entity_type`	`Mapped[str]`	The type of entity that changed, e.g. `"Task"`.
`entity_id`	`Mapped[str]`	UUID of the affected `Record`.
`action`	`Mapped[str]`	`"create"`, `"update"`, or `"delete"`.
`actor`	`Mapped[str \| None]`	Display name (or email) of the user who triggered the change, `None` for system-generated events.
`changes`	`Mapped[dict]`	Snapshot of the fields that were written (may be the full data dict on create, or just the changed keys on update).
`created_at`	`Mapped[datetime]`	UTC timestamp when the event was recorded.

Attributes¶

id `class-attribute` `instance-attribute` ¶

id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid4()))

entity_type `class-attribute` `instance-attribute` ¶

entity_type: Mapped[str] = mapped_column(String(64), index=True)

entity_id `class-attribute` `instance-attribute` ¶

entity_id: Mapped[str] = mapped_column(String(36), index=True)

action `class-attribute` `instance-attribute` ¶

action: Mapped[str] = mapped_column(String(16))

actor `class-attribute` `instance-attribute` ¶

actor: Mapped[str | None] = mapped_column(String(36))

changes `class-attribute` `instance-attribute` ¶

changes: Mapped[dict] = mapped_column(JSON, default=dict)

created_at `class-attribute` `instance-attribute` ¶

created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: now(UTC), index=True)

Events¶

bus ¶

In-process pub/sub event bus.

Callbacks registered with subscribe are called synchronously by emit. The events router uses this to push EventLog entries to WebSocket clients in real time. Exceptions in callbacks are swallowed so one bad subscriber cannot disrupt others.

Functions:

Name	Description
`subscribe`	Register a callback to receive future events.
`unsubscribe`	Remove a previously registered callback.
`emit`	Broadcast an event dict to all registered subscribers.

Functions:¶

subscribe ¶

subscribe(fn: Callable[[dict], None]) -> None

Register a callback to receive future events.

Parameters:

Name	Type	Description	Default
`fn`	`Callable[[dict], None]`	A callable that accepts one event dict argument.	required

Examples:

>>> events = []
>>> subscribe(events.append)
>>> emit({"action": "create"})
>>> events
[{'action': 'create'}]

unsubscribe ¶

unsubscribe(fn: Callable[[dict], None]) -> None

Remove a previously registered callback.

No-ops if fn is not in the subscriber list.

Parameters:

Name	Type	Description	Default
`fn`	`Callable[[dict], None]`	The callable to remove.	required

Examples:

>>> def handler(e): pass
>>> subscribe(handler)
>>> unsubscribe(handler)
>>> unsubscribe(handler)  # second call is a no-op

emit ¶

emit(event: dict) -> None

Broadcast an event dict to all registered subscribers.

Iterates over a snapshot of the subscriber list so that callbacks which call unsubscribe during iteration are safe. Exceptions raised by individual callbacks are silently swallowed.

Parameters:

Name	Type	Description	Default
`event`	`dict`	Arbitrary event payload dict.	required

Examples:

>>> received = []
>>> subscribe(received.append)
>>> emit({"entity_type": "Task", "action": "update"})
>>> received[0]["action"]
'update'

Media¶

storage ¶

Local filesystem media store.

Keys are relative path strings (e.g. "thumbnails/abc123.jpg"). All operations are relative to the configured root directory. Intermediate directories are created automatically on write.

Classes:

Name	Description
`LocalStorage`	Simple key-based local file store backed by a root directory.

Classes¶

LocalStorage ¶

LocalStorage(root: str)

Simple key-based local file store backed by a root directory.

Attributes:

Name	Type	Description
`root`		Absolute or relative `Path` to the storage root.

Examples:

>>> import tempfile, pathlib
>>> with tempfile.TemporaryDirectory() as tmp:
...     s = LocalStorage(tmp)
...     key = s.put("test/hello.txt", b"hi")
...     s.exists("test/hello.txt")
True

Initialise storage at root.

Parameters:

Name	Type	Description	Default
`root`	`str`	Path to the directory where files are stored. Created lazily when the first file is written.	required

Methods:

Name	Description
`fs_path`	Return the absolute filesystem path for a key.
`put`	Write raw bytes to a key, creating parent directories as needed.
`put_file`	Copy a local file into the store under `key`.
`exists`	Return `True` if the key resolves to an existing file.
`open`	Return a binary read handle for the stored file.

Attributes¶

root `instance-attribute` ¶

root = Path(root)

Methods:¶

fs_path ¶

fs_path(key: str) -> str

Return the absolute filesystem path for a key.

Parameters:

Name	Type	Description	Default
`key`	`str`	Storage key (relative path).	required

Returns:

Type	Description
`str`	Absolute path string.

Raises:

Type	Description
`ValueError`	If the key escapes the storage root.

Examples:

>>> import tempfile
>>> with tempfile.TemporaryDirectory() as tmp:
...     s = LocalStorage(tmp)
...     s.fs_path("a/b.jpg").endswith("a/b.jpg") or s.fs_path("a/b.jpg").endswith("a\\b.jpg")
True

put ¶

put(key: str, data: bytes) -> str

Write raw bytes to a key, creating parent directories as needed.

Parameters:

Name	Type	Description	Default
`key`	`str`	Target storage key.	required
`data`	`bytes`	Bytes to write.	required

Returns:

Type	Description
`str`	The key (for chaining / assignment).

Examples:

>>> import tempfile
>>> with tempfile.TemporaryDirectory() as tmp:
...     s = LocalStorage(tmp)
...     s.put("dir/file.bin", b"data")
'dir/file.bin'

put_file ¶

put_file(key: str, src_path: str) -> str

Copy a local file into the store under key.

Parameters:

Name	Type	Description	Default
`key`	`str`	Target storage key.	required
`src_path`	`str`	Absolute or relative path of the source file.	required

Returns:

Type	Description
`str`	The key.

Examples:

>>> put_file("thumb.jpg", "/tmp/generated.jpg")

exists ¶

exists(key: str) -> bool

Return True if the key resolves to an existing file.

Parameters:

Name	Type	Description	Default
`key`	`str`	Storage key to test.	required

Returns:

Type	Description
`bool`	`True` if the file exists, `False` otherwise.

Examples:

>>> import tempfile
>>> with tempfile.TemporaryDirectory() as tmp:
...     s = LocalStorage(tmp)
...     _ = s.put("x.bin", b"y")
...     s.exists("x.bin")
True

open ¶

open(key: str) -> BinaryIO

Return a binary read handle for the stored file.

Parameters:

Name	Type	Description	Default
`key`	`str`	Storage key of an existing file.	required

Returns:

Type	Description
`BinaryIO`	An open binary file object.

Raises:

Type	Description
`FileNotFoundError`	If the key does not exist.

Examples:

>>> storage.open("thumbnails/abc.jpg")

transcode ¶

ffmpeg-based thumbnail and proxy generation.

All functions require ffmpeg to be on PATH. Use has_ffmpeg() to check availability before calling. Failures raise subprocess.CalledProcessError.

Functions:

Name	Description
`has_ffmpeg`	Return `True` if `ffmpeg` is available on `PATH`.
`make_thumbnail`	Extract the "best" frame from a video or image and write a JPEG thumbnail.
`downscale_image`	Write a downscaled JPEG of a still image (never upscales) — used so an
`probe_duration`	Return a media file's duration in seconds (0.0 if unknown).
`probe_media`	Return basic media metadata for `src` via `ffprobe`.
`make_filmstrip`	Write a horizontal filmstrip sprite of `frames` evenly-spaced frames.
`extract_frame`	Extract a single frame (by index) from a video as a JPEG.
`make_proxy`	Transcode any ffmpeg-readable file to a browser-compatible H.264 MP4.

Functions:¶

has_ffmpeg ¶

has_ffmpeg() -> bool

Return True if ffmpeg is available on PATH.

Examples:

>>> isinstance(has_ffmpeg(), bool)
True

make_thumbnail ¶

make_thumbnail(src: str, dst: str, width: int = 320) -> str

Extract the "best" frame from a video or image and write a JPEG thumbnail.

Uses ffmpeg's thumbnail filter to pick the representative frame, then scales to width pixels (preserving aspect ratio).

Parameters:

Name	Type	Description	Default
`src`	`str`	Path to the source file (any ffmpeg-readable format).	required
`dst`	`str`	Destination path for the JPEG thumbnail.	required
`width`	`int`	Thumbnail width in pixels.	`320`

Returns:

Type	Description
`str`	`dst` on success.

Raises:

Type	Description
`CalledProcessError`	If ffmpeg exits with a non-zero code.

Examples:

>>> make_thumbnail("/input/video.mp4", "/out/thumb.jpg")

downscale_image ¶

downscale_image(src: str, dst: str, width: int = 512) -> str

Write a downscaled JPEG of a still image (never upscales) — used so an object thumbnail is a small image, not the user's full-resolution upload.

probe_duration ¶

probe_duration(src: str) -> float

Return a media file's duration in seconds (0.0 if unknown).

Uses ffprobe; returns 0.0 for stills or when probing fails, so callers can branch on duration > 0 to detect video.

Parameters:

Name	Type	Description	Default
`src`	`str`	Path to the source file.	required

Returns:

Type	Description
`float`	Duration in seconds, or `0.0`.

Examples:

>>> probe_duration("/in/clip.mp4")
2.0

probe_media ¶

probe_media(src: str) -> dict

Return basic media metadata for src via ffprobe.

Reads the first video stream's geometry / frame rate and the container duration. Missing or unprobeable values are simply omitted, so callers get a partial dict rather than an exception (a still image, for instance, has a width/height but no real duration).

Returns:

Type	Description
`dict`	A dict with any of `width`, `height` (ints), `duration` (seconds,
`dict`	float) and `fps` (float, rounded to 3 places). Empty on failure.

Examples:

>>> probe_media("/in/clip.mp4")
{'width': 1920, 'height': 1080, 'duration': 4.0, 'fps': 24.0}

make_filmstrip ¶

make_filmstrip(src: str, dst: str, frames: int = 12, width: int = 320) -> str

Write a horizontal filmstrip sprite of frames evenly-spaced frames.

Samples the clip at frames / duration fps and tiles the result into a single frames-column JPEG, so the web grid can scrub a Version by mapping the hover position to a column. Short clips simply leave trailing columns black (the sprite is always frames columns wide).

Parameters:

Name	Type	Description	Default
`src`	`str`	Path to the source video.	required
`dst`	`str`	Destination path for the JPEG sprite.	required
`frames`	`int`	Number of columns in the sprite.	`12`
`width`	`int`	Per-frame width in pixels (sprite width is `frames * width`).	`320`

Returns:

Type	Description
`str`	`dst` on success.

Raises:

Type	Description
`CalledProcessError`	If ffmpeg fails.
`ValueError`	If `src` has no duration (not a video).

Examples:

>>> make_filmstrip("/in/clip.mp4", "/out/strip.jpg")

extract_frame ¶

extract_frame(src: str, dst: str, frame: int, fps: float = 24.0) -> str

Extract a single frame (by index) from a video as a JPEG.

Seeks to (frame + 0.5) / fps seconds — the mid-frame, matching the player's seek math — and writes one frame. Used to render the exact annotated frame for the PDF review report.

Parameters:

Name	Type	Description	Default
`src`	`str`	Path to the source video.	required
`dst`	`str`	Destination image path.	required
`frame`	`int`	Zero-based frame index.	required
`fps`	`float`	Frames per second.	`24.0`

Returns:

Type	Description
`str`	`dst` on success.

Raises:

Type	Description
`CalledProcessError`	If ffmpeg fails.

Examples:

>>> extract_frame("/in/clip.mp4", "/out/f5.jpg", 5, 24)

make_proxy ¶

make_proxy(src: str, dst: str) -> str

Transcode any ffmpeg-readable file to a browser-compatible H.264 MP4.

-movflags +faststart moves the MOOV atom to the front of the file so the browser can begin playback before the download completes. Audio is stripped (-an) to keep the proxy small.

Parameters:

Name	Type	Description	Default
`src`	`str`	Path to the source file.	required
`dst`	`str`	Destination path for the MP4 proxy.	required

Returns:

Type	Description
`str`	`dst` on success.

Raises:

Type	Description
`CalledProcessError`	If ffmpeg fails.

Examples:

>>> make_proxy("/input/render.exr", "/out/proxy.mp4")

Python core - fxq¶

Application¶

app ¶

Classes¶

Functions:¶

create_app ¶

Configuration¶

config ¶

Classes¶

Settings ¶

Attributes¶

model_config class-attribute instance-attribute ¶

database_url class-attribute instance-attribute ¶

redis_url class-attribute instance-attribute ¶

s3_endpoint_url class-attribute instance-attribute ¶

s3_bucket class-attribute instance-attribute ¶

media_root class-attribute instance-attribute ¶

secret class-attribute instance-attribute ¶

cors_origins class-attribute instance-attribute ¶

debug class-attribute instance-attribute ¶

Functions:¶

get_settings ¶

Database¶

db ¶

Classes¶

Base ¶

Functions:¶

make_engine ¶

make_session_factory ¶

Schema engine¶

types ¶

Attributes¶

LINK_TYPES module-attribute ¶

Classes¶

FieldDefn dataclass ¶

Attributes¶

code instance-attribute ¶

display_name instance-attribute ¶

data_type instance-attribute ¶

config class-attribute instance-attribute ¶

is_required class-attribute instance-attribute ¶

inherit class-attribute instance-attribute ¶

is_link property ¶

link_types property ¶

EntityDefn dataclass ¶

Attributes¶

code instance-attribute ¶

display_name instance-attribute ¶

hierarchy_mode class-attribute instance-attribute ¶

parent_field class-attribute instance-attribute ¶

fields class-attribute instance-attribute ¶

Methods:¶

field ¶

has_field ¶

Schema dataclass ¶

Attributes¶

entities class-attribute instance-attribute ¶

Methods:¶

entity ¶

has_entity ¶

loader ¶

Classes¶

Functions:¶

load_schema ¶

Query DSL¶

dsl ¶

Attributes¶

OPERATORS module-attribute ¶

Classes¶

DslError ¶

Condition dataclass ¶

Attributes¶

path instance-attribute ¶

op instance-attribute ¶

value instance-attribute ¶

Group dataclass ¶

Attributes¶

op instance-attribute ¶

children instance-attribute ¶

Functions:¶

Python core - `fxq`¶

model_config `class-attribute` `instance-attribute` ¶

database_url `class-attribute` `instance-attribute` ¶

redis_url `class-attribute` `instance-attribute` ¶

s3_endpoint_url `class-attribute` `instance-attribute` ¶

s3_bucket `class-attribute` `instance-attribute` ¶

media_root `class-attribute` `instance-attribute` ¶

secret `class-attribute` `instance-attribute` ¶

cors_origins `class-attribute` `instance-attribute` ¶

debug `class-attribute` `instance-attribute` ¶

LINK_TYPES `module-attribute` ¶

FieldDefn `dataclass` ¶

code `instance-attribute` ¶

display_name `instance-attribute` ¶

data_type `instance-attribute` ¶

config `class-attribute` `instance-attribute` ¶

is_required `class-attribute` `instance-attribute` ¶

inherit `class-attribute` `instance-attribute` ¶

is_link `property` ¶

link_types `property` ¶

EntityDefn `dataclass` ¶

code `instance-attribute` ¶

display_name `instance-attribute` ¶

hierarchy_mode `class-attribute` `instance-attribute` ¶

parent_field `class-attribute` `instance-attribute` ¶

fields `class-attribute` `instance-attribute` ¶

Schema `dataclass` ¶

entities `class-attribute` `instance-attribute` ¶

OPERATORS `module-attribute` ¶

Condition `dataclass` ¶

path `instance-attribute` ¶

op `instance-attribute` ¶

value `instance-attribute` ¶

Group `dataclass` ¶

op `instance-attribute` ¶

children `instance-attribute` ¶

id `class-attribute` `instance-attribute` ¶

entity_type `class-attribute` `instance-attribute` ¶

project_id `class-attribute` `instance-attribute` ¶

status `class-attribute` `instance-attribute` ¶

data `class-attribute` `instance-attribute` ¶

active `class-attribute` `instance-attribute` ¶

thumbnail_id `class-attribute` `instance-attribute` ¶

created_by `class-attribute` `instance-attribute` ¶

created_at `class-attribute` `instance-attribute` ¶

updated_at `class-attribute` `instance-attribute` ¶

id `class-attribute` `instance-attribute` ¶

code `class-attribute` `instance-attribute` ¶

display_name `class-attribute` `instance-attribute` ¶

is_builtin `class-attribute` `instance-attribute` ¶

icon `class-attribute` `instance-attribute` ¶

hierarchy_mode `class-attribute` `instance-attribute` ¶

parent_field `class-attribute` `instance-attribute` ¶

fields `class-attribute` `instance-attribute` ¶

id `class-attribute` `instance-attribute` ¶

entity_type_id `class-attribute` `instance-attribute` ¶

code `class-attribute` `instance-attribute` ¶

display_name `class-attribute` `instance-attribute` ¶

data_type `class-attribute` `instance-attribute` ¶

config `class-attribute` `instance-attribute` ¶

is_builtin `class-attribute` `instance-attribute` ¶

is_required `class-attribute` `instance-attribute` ¶

sort_order `class-attribute` `instance-attribute` ¶

default `class-attribute` `instance-attribute` ¶

inherit `class-attribute` `instance-attribute` ¶

entity_type `class-attribute` `instance-attribute` ¶

id `class-attribute` `instance-attribute` ¶

code `class-attribute` `instance-attribute` ¶

name `class-attribute` `instance-attribute` ¶

color `class-attribute` `instance-attribute` ¶

icon `class-attribute` `instance-attribute` ¶