Python API

Every name ceres exports, grouped by what it is for. Each is importable straight from the package, from ceres import Component.

Components

Bound

Type-only marker used as Bound[Connection] to declare a field bound to a sibling object.

Bound fields are how a component declares it expects an external object (currently only connections) to be supplied through the BoundField annotation. The actual descriptor behaviour is implemented by BoundField, this class only exists to give type checkers a handle.

Component

Bases: DataObject, ComponentSource

Primary unit of organization in Ceres.

A Component is a Pydantic-style dataclass that bundles configurable state with decorated methods (@listener, @routine, @query, @action, @sieve) and child components. Each driver, simulator, sieve container, etc. is a Component.

Components are arranged in a tree, each with an address like @parent.child. The runtime behaviour (lifecycle, event bus, references) lives on the paired ComponentSystem accessible via system. The component class itself stays focused on declarative state and the methods the user writes.

Subclasses typically override __setup__, __start__, __stop__, and the __static_connections__ / __static_sieves__ / __static_jobs__ / __static_pruners__ hooks to declare static behaviour, or define methods decorated with @listener, @routine, @query, @action, and @sieve.

system property

system: ComponentSystem

The runtime system that owns this component's lifecycle, events, and managers.

__bind__

__bind__(bind: ComponentSystem) -> None

Re-bind this component to a different ComponentSystem.

Used internally when a system is constructed for a component, or when the engine swaps a component into a fresh system during reconfiguration.

__connectivity__

__connectivity__() -> Connectivity | None

Return the component's connectivity status, or None if not applicable.

__particles__ classmethod

__particles__() -> tuple[type[Particle], ...]

Particle classes this component declares, offered to chart pickers.

Computed from @sieve-decorated method return annotations. Assign a literal tuple instead to declare particles emitted outside any sieve.

__setup__

__setup__() -> None

Hook called after the component is constructed and bound to its system.

Override to perform initialization that depends on self.system being available.

__start__

__start__() -> None | Awaitable[None]

Hook called when the component starts.

Override to perform startup work. May be synchronous or return an awaitable.

__static_connections__

__static_connections__() -> Iterable[ConnectionConfig]

Return connection configurations declared statically by this component class.

__static_jobs__

__static_jobs__() -> Iterable[JobConfig]

Return job configurations declared statically by this component class.

__static_pruners__

__static_pruners__() -> Iterable[PrunerConfig]

Return pruner configurations declared statically by this component class.

__static_sieves__

__static_sieves__() -> Iterable[SieveConfig]

Return sieve configurations declared statically by this component class.

__stop__

__stop__() -> None | Awaitable[None]

Hook called when the component stops.

Override to release resources. May be synchronous or return an awaitable. Always called even if __start__ raised.

ComponentAccessLevel

Bases: OrderedStrEnum

Access level controlling what a user may do with a component.

The hierarchy is strict: each level implies all levels below it. DENY is only valid as a default access level on a component definition and means no access unless explicitly granted.

Defined here rather than in ceres.component because ComponentConfig needs it as a field type and ceres.component already imports from this module at load time. ceres.component re-exports this enum for API discoverability.

DENY class-attribute instance-attribute

DENY = 'deny'

No access, the component is invisible to the user.

MANAGE class-attribute instance-attribute

MANAGE = 'manage'

Can change configuration and manage permissions.

OPERATE class-attribute instance-attribute

OPERATE = 'operate'

Can invoke actions and send data on connections.

VIEW class-attribute instance-attribute

VIEW = 'view'

Can see the component and view its data.

ComponentSystem

Bases: Node, ComponentSource

Runtime wrapper around a Component that owns its lifecycle and infrastructure.

Each Component is paired with exactly one ComponentSystem. The system provides:

• Tree placement via the parent container and child registry.
• Lifecycle: start, stop, enable, disable, plus the routine and procedure runners.
• The event bus and reference resolution.
• Lazy access to the connection, sieve, job, and pruner managers.

Components reach their system via Component.system. The system is normally created by Component.__post_init__, callers rarely instantiate it directly.

access property

access: ComponentAccessLevel | None

Default access level from this component's config, or None if not set.

address property

address: Address

The current address of the component, derived by walking up to its top-level ancestor.

children property

children: Sequence[ComponentSystem]

All child component systems of this component, in registered order.

component property

component: Component

The underlying component this system wraps.

config property writable

config: ComponentConfig | None

The configuration of the component, if available.

connections cached property

connections: ComponentConnectionManager

Manager for the component's network connections.

container property writable

container: ComponentSystem | Engine | None

The container of the component system.

This is either another ComponentSystem (the parent component) or an Engine (when this component is a top-level component). Returns None when the component is detached.

database property

database: Database

The database used by the component.

If the component is part of a tree, this returns the database from the engine at the root. A detached component creates a private temporary database lazily on first access, useful for unit tests. Its files are removed when the database is disposed.

enabled property

enabled: bool

True if the component is enabled.

Enabled components start automatically when their parent or containing engine starts.

engine property

engine: Engine | None

The engine that contains this component, or None if it is detached.

Walks the container chain upwards and returns the engine at the top.

jobs cached property

jobs: JobManager

Manager for scheduled jobs declared by this component.

name property writable

name: Name

The name this component is registered under within its parent.

parent property

parent: ComponentSystem | None

The parent component's system, or None if there is no parent.

pruners cached property

pruners: PrunerManager

Manager for the component's data pruners.

root property

root: ComponentSystem

The top-level system of this component's tree, or itself when it has no parent.

sieves cached property

sieves: SieveManager

Manager for the component's sieves.

tags property

tags: list[str]

Tags declared on this component's config, empty if none.

attach

attach(
    child: Component | ComponentSystem,
    /,
    name: Name | None = None,
) -> None

Add a child component beneath this component.

Parameters:

Name	Type	Description	Default
child	Component | ComponentSystem	Component or component system to attach.	required
name	Name | None	Optional name to register the child under. When omitted, the child's existing name is used.	None

Raises:

Type	Description
ValueError	If the child is this component itself, an ancestor, or there is already another child registered under the same name.

call async

call(
    procedure: str,
    arguments: Mapping[Name, Any] | None = None,
) -> object | None

Invoke a procedure once and return its output.

For live procedures, queries return the first yielded value while actions exhaust the async iterator and return the last one.

Parameters:

Name	Type	Description	Default
procedure	str	Name of the procedure to invoke.	required
arguments	Mapping[Name, Any] | None	Mapping of argument names to values, or None for no arguments.	None

Returns:

Type	Description
object | None	The procedure's output. BaseOutput instances are returned as-is, plain values are
object | None	returned as the procedure produced them.

Raises:

Type	Description
ProcedureNotFoundError	If no procedure exists with the given name.
ProcedureInvalidArgumentsError	If the procedure's arguments fail validation.
ProcedureInternalError	If the procedure raises an exception during execution.

contains

contains(
    component: Component | ComponentSystem,
    *,
    inclusive: bool = False
) -> bool

Return True if component is somewhere in this subtree.

Parameters:

Name	Type	Description	Default
component	Component | ComponentSystem	Component or system to test.	required
inclusive	bool	When True, return True for the component itself in addition to its descendants.	False

detach

detach() -> None

Remove the component from its container.

The container can be either a parent component or the engine. If the component has no container, this does nothing. Emits WillDetachEvent before the removal and DetachedEvent afterward (propagated through the former container so listeners can see what was detached and from where).

disable async

disable() -> None

Disable the component and persist the new state.

Disabled components do not start automatically when their parent or containing engine starts. Emits DisabledEvent on success.

down async

down() -> None

Disable and stop the component, waiting for it to fully stop.

enable async

enable() -> None

Enable the component and persist the new state.

Enabled components start automatically when their parent or containing engine starts. Emits EnabledEvent on success.

get_action_bindings

get_action_bindings() -> Mapping[Name, ActionBinding]

Return a mapping of action names to action bindings on the underlying class.

get_ancestor_components

get_ancestor_components(
    *, inclusive: bool = False
) -> list[Component]

Return a list of ancestor components in ascending order.

Parameters:

Name	Type	Description	Default
inclusive	bool	When True, include this component itself as the first entry.	False

Returns:

Type	Description
list[Component]	Ancestors ordered from nearest parent to root.

get_component

get_component(
    address: str | DynamicAddress | None = None,
) -> Component | None

Resolve address to a component beneath this one.

Parameters:

Name	Type	Description	Default
address	str | DynamicAddress | None	Address string or DynamicAddress. An empty value returns this component. Absolute addresses route through the engine when this system is attached to one so they can reach components in any top-level tree. On a detached system, absolute addresses are resolved against the local tree root instead.	None

Returns:

Type	Description
Component | None	The matching component, or None if no component exists at that address.

get_components

get_components(
    filter: ComponentFilter | AddressSelector | None = None,
    /,
    *,
    inclusive: bool = False,
    **kwargs: Unpack[ComponentFilterArgs],
) -> list[Component]

Walk this component's subtree and return components matching the given filter.

Parameters:

Name	Type	Description	Default
filter	ComponentFilter | AddressSelector | None	A ComponentFilter or AddressSelector, or None to skip positional filtering.	None
inclusive	bool	When True, include this component itself in the candidate set.	False
**kwargs	Unpack[ComponentFilterArgs]	Additional filter overrides.	{}

Returns:

Type	Description
list[Component]	All matching components in pre-order traversal.

get_connection_bindings

get_connection_bindings() -> (
    Mapping[Name, ConnectionBinding]
)

Return a mapping of connection names to connection bindings on the underlying class.

get_inherited_tags

get_inherited_tags() -> set[str]

Collect tags from this component, all ancestors, and the config level.

get_listener_bindings

get_listener_bindings() -> Sequence[ListenerBinding]

Return every listener binding on the underlying component class.

get_procedure_bindings

get_procedure_bindings() -> Mapping[Name, ProcedureBinding]

Return a mapping of procedure names to bindings (queries plus actions).

get_query_bindings

get_query_bindings() -> Mapping[Name, QueryBinding]

Return a mapping of query names to query bindings on the underlying class.

get_referenced_components

get_referenced_components(
    reference: str | None = None,
) -> list[Component]

Return components reached by walking this component's references.

Parameters:

Name	Type	Description	Default
reference	str | None	Optional dotted attribute path to start the walk from. When given, only the subtree rooted at that attribute is traversed.	None

Returns:

Type	Description
list[Component]	All other components that this component (or the requested subtree) references.
list[Component]	Self-references are excluded.

get_references

get_references() -> list[Reference]

Walk the component's state and collect every Reference instance found.

get_referencing_components

get_referencing_components(
    recursive: bool = False,
) -> list[Component]

Return components that hold a reference to this component.

Parameters:

Name	Type	Description	Default
recursive	bool	When True, also include components that reference this component indirectly through a chain of other referencers.	False

Returns:

Type	Description
list[Component]	The list of referencing components.

get_resolved_access

get_resolved_access() -> ComponentAccessLevel

Walk the ancestor chain to find the nearest declared access level.

Return the first non-None access found walking from this component up to its top-level ancestor, then the config-level default, then ComponentAccessLevel.VIEW.

get_routine_bindings

get_routine_bindings() -> Sequence[RoutineBinding]

Return every routine binding on the underlying component class.

get_sieve_bindings

get_sieve_bindings() -> Mapping[Name, SieveBinding]

Return a mapping of sieve names to sieve bindings on the underlying class.

get_status async

get_status() -> Status

Return the component's status augmented with its enabled flag and connectivity.

A component may define __connectivity__ to report a single overall connectivity state. Otherwise the per-connection states are reported so observers can represent each one.

has_reference_to

has_reference_to(component: Component) -> bool

Return True if this component references component directly.

start

start(
    *,
    on_completed: Callable[[Self], None] | None = None,
    on_exception: (
        Callable[[Self, BaseException], None] | None
    ) = None,
    all_enabled: bool = True
) -> None

Start the component, its ancestors, and any enabled descendants.

Ancestors are started first so this component always has a running parent. After this component starts, every enabled child is started recursively unless all_enabled is False.

Parameters:

Name	Type	Description	Default
on_completed	Callable[[Self], None] | None	Optional callback invoked when the component task finishes successfully.	None
on_exception	Callable[[Self, BaseException], None] | None	Optional callback invoked when the component task fails.	None
all_enabled	bool	When True, recursively start every enabled descendant. Internal callers set this to False when starting an ancestor to avoid double-starts.	True

subscribe async

subscribe(
    procedure: str,
    arguments: Mapping[Name, Any] | None = None,
) -> AsyncIterable[object | None]

Subscribe to a procedure, yielding its output as it becomes available.

For non-live queries, this polls the procedure on the binding's poll interval and yields each result. For live procedures, this yields each value the underlying async generator produces.

Parameters:

Name	Type	Description	Default
procedure	str	Name of the procedure to subscribe to.	required
arguments	Mapping[Name, Any] | None	Mapping of argument names to values, or None for no arguments.	None

Yields:

Type	Description
AsyncIterable[object | None]	Each successive output value from the procedure.

Raises:

Type	Description
ProcedureNotFoundError	If no procedure exists with the given name.
ProcedureNotSubscribableError	If the procedure is a non-live action, which cannot be subscribed to.
ProcedureInternalError	If the procedure raises an exception during execution.

sync_child_order

sync_child_order() -> None

Reorder the child registry to match the order specified in the component's config.

Children present in the configuration come first in declared order, any extra children not mentioned in the configuration are appended afterward in their existing order.

sync_references

sync_references() -> (
    tuple[list[Reference], list[Reference]]
)

Resolve every Reference declared by the component against the current tree.

Each reference's root is set to this component, then resolution is attempted. Components whose references resolve are added to the target's referencer set so the target knows who depends on it. Stale referencers (where this system no longer references them) are removed.

Returns:

Type	Description
tuple[list[Reference], list[Reference]]	A (resolved, unresolved) tuple of references that did and did not resolve.

up async

up() -> None

Enable and start the component.

FileOutput

Bases: BaseOutput

Procedure output that streams a file from disk as the HTTP response.

__init__

__init__(
    path: str | PathLike,
    media: OutputMediaType | None = None,
    *,
    http_status: int = 200,
    http_headers: Mapping[str, str] | None = None,
    http_filename: str | None = None,
    on_exit: Callable[[], Awaitable[Any]] | None = None
) -> None

Construct a file output.

Parameters:

Name	Type	Description	Default
path	str | PathLike	Path to the file to send.	required
media	OutputMediaType | None	Optional MIME type, inferred from the path when omitted.	None
http_status	int	HTTP status code for the response.	200
http_headers	Mapping[str, str] | None	Additional response headers.	None
http_filename	str | None	Filename hint to send via the Content-Disposition header.	None
on_exit	Callable[[], Awaitable[Any]] | None	Optional async callback run after the response finishes streaming.	None

ProcedureType

Bases: StrEnum

Discriminator for the two RPC-style procedures, queries and actions.

ACTION class-attribute instance-attribute

ACTION = 'action'

Mutating procedure that performs side effects.

QUERY class-attribute instance-attribute

QUERY = 'query'

Read-only procedure that returns a value derived from component state.

StreamingOutput

Bases: BaseOutput

Procedure output that streams arbitrary bytes as the HTTP response body.

Outside a response, read the bytes through the output as an async context manager:

async with output:
    async for chunk in output:
        ...

Leaving the block closes the stream, so the producer's cleanup runs even after an early break, and then awaits on_exit. A factory stream opens again on the next async with, while a plain async iterable is spent after one pass.

Returned from a procedure, the output ends that procedure call when it first closes, with ProcedureCompletedEvent when the stream ran out, ProcedureExceptionEvent when reading it raised, and ProcedureCancelledEvent when it closed before its end. That holds whoever reads it, the server sending it as a response or Python code in an async with block. A returned output that is never opened never ends its call, so no end event follows its ProcedureCalledEvent.

__aenter__ async

__aenter__() -> Self

Open the stream, calling a factory stream to start its iterable.

Raises:

Type	Description
RuntimeError	If the output is already open.

__aexit__ async

__aexit__(*exc_info: object) -> None

Close the stream so its producer's cleanup runs, then await on_exit.

A first close of an output a procedure returned also reports how that call ended.

__aiter__

__aiter__() -> AsyncIterator[DataStreamChunk]

Return the open stream's chunks.

Raises:

Type	Description
RuntimeError	If the output is not open.

__init__

__init__(
    stream: DataStream,
    media: OutputMediaType,
    *,
    http_status: int = 200,
    http_headers: Mapping[str, str] | None = None,
    http_filename: str | None = None,
    on_exit: Callable[[], Awaitable[Any]] | None = None
) -> None

Construct a streaming output.

Parameters:

Name	Type	Description	Default
stream	DataStream	Async iterable of byte chunks, or a zero-arg factory returning one. A factory starts a fresh iterable each time the output opens.	required
media	OutputMediaType	MIME type to advertise on the response.	required
http_status	int	HTTP status code for the response.	200
http_headers	Mapping[str, str] | None	Additional response headers.	None
http_filename	str | None	Filename hint to send via the Content-Disposition header.	None
on_exit	Callable[[], Awaitable[Any]] | None	Optional async callback run after the output closes.	None

action

action(method: Callable[P, T]) -> Callable[P, T]

action(
    *,
    media: str | None = None,
    permit: ProcedurePermissionsInput = ...
) -> Callable[[Callable[P, T]], Callable[P, T]]

action(
    method: Callable[P, T] | None = None,
    *,
    media: str | None = None,
    permit: ProcedurePermissionsInput = ComponentAccessLevel.OPERATE
) -> (
    Callable[P, T]
    | Callable[[Callable[P, T]], Callable[P, T]]
)

Mark a method as an action, a mutating RPC endpoint that performs side effects.

Like @query, actions may return JSON-serializable values, file or streaming outputs, or yield live values from an async generator.

Parameters:

Name	Type	Description	Default
method	Callable[P, T] | None	When used without arguments, the method being decorated.	None
media	str | None	Media type for streaming actions, required when the return type is StreamingOutput and otherwise optional.	None
permit	ProcedurePermissionsInput	Minimum access level required to invoke the action, or "public" to allow unauthenticated access.	OPERATE

Returns:

Type	Description
Callable[P, T] | Callable[[Callable[P, T]], Callable[P, T]]	Either the decorated method (when used without arguments) or a decorator returning the
Callable[P, T] | Callable[[Callable[P, T]], Callable[P, T]]	decorated method.

listener

listener(method: _ListenerMethod) -> _ListenerMethod

listener(
    *,
    event: type | UnionType | None = None,
    local: bool | None = None,
    reference: str | Sequence[str] | None = None,
    address: (
        str
        | AddressSelector
        | Sequence[str | AddressSelector]
        | None
    ) = None
) -> _ListenerMethodTransform

listener(
    method: _ListenerMethod | None = None,
    *,
    event: type | UnionType | None = None,
    local: bool | None = None,
    reference: str | Sequence[str] | None = None,
    address: (
        str
        | AddressSelector
        | Sequence[str | AddressSelector]
        | None
    ) = None
) -> _ListenerMethod | _ListenerMethodTransform

Mark a method as a listener that runs in response to events on the event bus.

The decorated method may take just self, or self plus the event instance. The event type is inferred from the second parameter's type hint when event is not given explicitly, falling back to the base Event so the listener fires on every event.

Parameters:

Name	Type	Description	Default
method	_ListenerMethod | None	When used without arguments, the method being decorated.	None
event	type | UnionType | None	Specific event type (or union of types) to listen for. If omitted, inferred from the method's second parameter type hint.	None
local	bool | None	When True, only events emitted on the listener's own component are delivered. Defaults to True when neither reference nor address is given, otherwise False.	None
reference	str | Sequence[str] | None	One or more dotted reference paths (relative to the component) whose targets' events should also be delivered.	None
address	str | AddressSelector | Sequence[str | AddressSelector] | None	One or more address selectors identifying additional components whose events should be delivered.	None

Returns:

Type	Description
_ListenerMethod | _ListenerMethodTransform	Either the decorated method (when used without arguments) or a decorator returning the
_ListenerMethod | _ListenerMethodTransform	decorated method.

query

query(method: Callable[P, T]) -> Callable[P, T]

query(
    *, media: str, permit: ProcedurePermissionsInput = ...
) -> Callable[[Callable[P, T]], Callable[P, T]]

query(
    *,
    poll: float | timedelta = ...,
    permit: ProcedurePermissionsInput = ...
) -> Callable[[Callable[P, T]], Callable[P, T]]

query(
    method: Callable[P, T] | None = None,
    *,
    poll: float | timedelta = timedelta(seconds=5),
    media: str | None = None,
    permit: ProcedurePermissionsInput = ComponentAccessLevel.VIEW
) -> (
    Callable[P, T]
    | Callable[[Callable[P, T]], Callable[P, T]]
)

Mark a method as a query, a read-only RPC endpoint with optional polling subscription.

Queries may be plain methods returning a JSON-serializable value, methods returning a FileOutput or StreamingOutput, or async generators yielding live values.

Parameters:

Name	Type	Description	Default
method	Callable[P, T] | None	When used without arguments, the method being decorated.	None
poll	float | timedelta	Interval used when subscribing to a non-live query.	timedelta(seconds=5)
media	str | None	Media type for streaming queries, required when the return type is StreamingOutput and otherwise optional.	None
permit	ProcedurePermissionsInput	Minimum access level required to call the query, or "public" to allow unauthenticated access.	VIEW

Returns:

Type	Description
Callable[P, T] | Callable[[Callable[P, T]], Callable[P, T]]	Either the decorated method (when used without arguments) or a decorator returning the
Callable[P, T] | Callable[[Callable[P, T]], Callable[P, T]]	decorated method.

routine

routine(
    *,
    restart: (
        RoutineRestartPolicy | RoutineRestartPolicyLiteral
    ) = RoutineRestartPolicy.NEVER,
    restart_delay: (
        PositiveFloat | PositiveTimeDelta
    ) = timedelta(seconds=1)
) -> _RoutineMethodHandler

routine(method: _RoutineMethod) -> _RoutineMethod

routine(
    method: _RoutineMethod | None = None,
    *,
    restart: (
        RoutineRestartPolicy | RoutineRestartPolicyLiteral
    ) = RoutineRestartPolicy.NEVER,
    restart_delay: (
        PositiveFloat | PositiveTimeDelta
    ) = timedelta(seconds=1)
) -> _RoutineMethod | _RoutineMethodHandler

Mark an async method as a routine, a long-running background task.

Routines are started when the component starts and cancelled when it stops. The restart policy controls whether the routine is re-run after it finishes, with restart_delay inserted between attempts.

Parameters:

Name	Type	Description	Default
method	_RoutineMethod | None	When used without arguments, the method being decorated.	None
restart	RoutineRestartPolicy | RoutineRestartPolicyLiteral	Restart policy. Defaults to NEVER, meaning the routine runs once.	NEVER
restart_delay	PositiveFloat | PositiveTimeDelta	Delay before restarting, accepted as either seconds or a timedelta.	timedelta(seconds=1)

Returns:

Type	Description
_RoutineMethod | _RoutineMethodHandler	Either the decorated method (when used without arguments) or a decorator returning the
_RoutineMethod | _RoutineMethodHandler	decorated method.

sieve

sieve(method: SieveMethod[S, T]) -> SieveMethod[S, T]

sieve(
    *connections: MaybeSequence[ConnectionField] | None,
    direction: (
        MaybeSequence[MessageDirectionInput] | None
    ) = "receive",
    name: Name | None = None,
    stored: bool = True,
    retries: NonNegativeInt | None = None,
    retry_delay: PositiveTimeDelta = timedelta(seconds=5),
    contains: MaybeSequence[MessageData] | None = None,
    prefix: MaybeSequence[MessageData] | None = None,
    suffix: MaybeSequence[MessageData] | None = None,
    buffer_size: int | str | None = None,
    buffer_drop: int | str | None = None
) -> Callable[[SieveMethod[S, T]], SieveMethod[S, T]]

sieve(
    *given: SieveMethod[S, T]
    | MaybeSequence[ConnectionField]
    | None,
    direction: (
        MaybeSequence[MessageDirectionInput] | None
    ) = "receive",
    name: Name | None = None,
    stored: bool = True,
    retries: NonNegativeInt | None = None,
    retry_delay: PositiveTimeDelta = timedelta(seconds=5),
    contains: MaybeSequence[MessageData] | None = None,
    prefix: MaybeSequence[MessageData] | None = None,
    suffix: MaybeSequence[MessageData] | None = None,
    buffer_size: int | str | None = None,
    buffer_drop: int | str | None = None
) -> (
    SieveMethod[S, T]
    | Callable[[SieveMethod[S, T]], SieveMethod[S, T]]
)

Mark a method as a sieve, parsing incoming messages into particles.

A sieve method takes either a single Message (returning an optional particle), an async iterable of messages (yielding particles), or a Buffer (yielding particles).

Parameters:

Name	Type	Description	Default
*given	SieveMethod[S, T] | MaybeSequence[ConnectionField] | None	Connections to bind the sieve to, each a connection or a sequence of them. When used as @sieve without arguments, this is the decorated method itself.	()
direction	MaybeSequence[MessageDirectionInput] | None	Message direction(s) to accept. Defaults to receive-only.	'receive'
name	Name | None	Optional override for the sieve name, defaults to the method name.	None
stored	bool	Whether produced particles should be persisted to the database.	True
retries	NonNegativeInt | None	Maximum retries for failed messages, or None for unlimited.	None
retry_delay	PositiveTimeDelta	Delay between retries.	timedelta(seconds=5)
contains	MaybeSequence[MessageData] | None	Filter accepting only messages whose data contains the given bytes.	None
prefix	MaybeSequence[MessageData] | None	Filter accepting only messages whose data starts with the given bytes.	None
suffix	MaybeSequence[MessageData] | None	Filter accepting only messages whose data ends with the given bytes.	None
buffer_size	int | str | None	Optional cap on the in-memory buffer this sieve receives.	None
buffer_drop	int | str | None	Optional threshold at which oldest data is dropped from the buffer.	None

Returns:

Type	Description
SieveMethod[S, T] | Callable[[SieveMethod[S, T]], SieveMethod[S, T]]	Either the decorated method (when used without arguments) or a decorator returning the
SieveMethod[S, T] | Callable[[SieveMethod[S, T]], SieveMethod[S, T]]	decorated method.

Engine

Bases: Node

Top-level container that owns the engine's components and the supporting infrastructure.

The engine is the entry point for a Ceres process. It owns the Database, the optional HTTP Server, and the top-level components. It loads configuration, reconciles the running components with the desired configuration, and drives the lifecycle (start/stop) of everything beneath it.

There is exactly one engine per Ceres process. Components reach it through ComponentSystem.engine.

address property

address: Address

The synthetic address used for the engine itself, always Address.engine().

config property

config: Config

The configuration most recently applied to the engine.

config_path property

config_path: Path | None

Filesystem path the current configuration was loaded from, if any.

database property

database: Database

Return the engine's database instance.

default_access property

default_access: ComponentAccessLevel | None

Config-level default access for components with none declared, or None.

default_tags property

default_tags: list[str]

Tags from the top-level configuration, inherited by every component.

engine property

engine: Self

Return this engine instance, satisfying the Node.engine interface.

project_directory property

project_directory: Directory | None

Directory containing the configuration file, or None if no path is known.

server property

server: Server | None

The currently running HTTP server, or None if no server is active.

settings cached property

settings: SettingManager

Manager for engine-wide settings.

users cached property

users: UserManager

Manager for user accounts.

workspace_edits cached property

workspace_edits: WorkspaceEditManager

Manager for in-progress workspace edits.

workspaces cached property

workspaces: WorkspaceManager

Manager for workspaces.

__manager__

__manager__(Entity: type[Entity]) -> BaseEntityManager

Return the entity manager bound to this engine for the given Entity type.

attach

attach(
    component: Component | ComponentSystem,
    /,
    name: Name | None = None,
) -> Component | None

Attach a component as a top-level component of the engine.

Parameters:

Name	Type	Description	Default
component	Component | ComponentSystem	The component or component system to attach.	required
name	Name | None	Optional name to assign to the component before attaching it.	None

Returns:

Type	Description
Component | None	The previously attached top-level component with the same name if one was
Component | None	replaced, otherwise None.

detach

detach(component: Component | ComponentSystem) -> None

Remove a top-level component from the engine.

get_component

get_component(
    address: str | DynamicAddress | None = None,
) -> Component | None

Look up a component by address anywhere in the component tree.

Parameters:

Name	Type	Description	Default
address	str | DynamicAddress | None	An address string or DynamicAddress. None or an empty value returns None, there is no default component.	None

Returns:

Type	Description
Component | None	The matching component, or None if nothing matches.

get_components

get_components(
    filter: ComponentFilter | AddressSelector | None = None,
    /,
    *,
    inclusive: bool = False,
    **kwargs: Unpack[ComponentFilterArgs],
) -> list[Component]

Walk every top-level component tree and return components matching the given filter.

Parameters:

Name	Type	Description	Default
filter	ComponentFilter | AddressSelector | None	A ComponentFilter or AddressSelector to apply, or None to skip positional filtering.	None
inclusive	bool	Accepted for interface compatibility with ComponentSystem.get_components, the engine always includes every top-level component regardless.	False
**kwargs	Unpack[ComponentFilterArgs]	Additional filter overrides forwarded as ComponentFilterArgs.	{}

Returns:

Type	Description
list[Component]	A list of matching components, or an empty list when the engine has no components.

hash_password async

hash_password(password: str) -> PasswordHash

Hash a plaintext password using the engine's database password hasher.

load async

load(
    source: ConfigSource[Config],
    *,
    checks: Sequence[
        ConfigCheckType
    ] = ConfigCheckType.all(),
    silent: bool = False
) -> Config

Load and apply a configuration from the given source.

Parameters:

Name	Type	Description	Default
source	ConfigSource[Config]	Path to a configuration file, or a Config instance, or any other ConfigSource[Config].	required
checks	Sequence[ConfigCheckType]	Configuration checks to run during loading.	all()
silent	bool	When True, suppress informational logging.	False

Returns:

Type	Description
Config	The fully resolved Config that was applied.

reload async

reload(
    *,
    checks: Sequence[
        ConfigCheckType
    ] = ConfigCheckType.all(),
    silent: bool = False
) -> Config

Re-load the current configuration from disk, falling back to the in-memory copy.

Parameters:

Name	Type	Description	Default
checks	Sequence[ConfigCheckType]	Configuration checks to run during loading.	all()
silent	bool	When True, suppress informational logging.	False

Returns:

Type	Description
Config	The freshly loaded Config that was applied.

Raises:

Type	Description
ReloadConfigInvalidError	If the configuration file fails validation. Wrap the underlying ConfigError so callers can distinguish reload-time validation problems from other errors.

verify_password async

verify_password(password: str, hash: PasswordHash) -> bool

Verify a plaintext password against a previously stored hash.

Loaded module-attribute

Loaded = _LoadedType

Loader

Bases: DataObject

Declarative specification for constructing an instance of a class by import path.

A Loader names a target class via its fully qualified import string and collects the arguments used to construct it. Calling create() imports the class and instantiates it, the validator also performs a dry-run construction so misconfiguration is caught at load time rather than at first use.

arguments class-attribute instance-attribute

arguments: Mapping[str, Any] = Field(default_factory=dict)

Keyword arguments passed when constructing the target class.

cls class-attribute instance-attribute

cls: ImportString[type[T]] = Field(
    validation_alias="class", serialization_alias="class"
)

Fully qualified import path of the target class.

create

create(arguments: Mapping[str, Any] | None = None) -> T

Instantiate the target class and return the resulting object.

Parameters:

Name	Type	Description	Default
arguments	Mapping[str, Any] | None	Additional keyword arguments that override any overlapping entries from arguments and the loader's extra kwargs.	None

Returns:

Type	Description
T	A new instance of the loader's target class.

ScheduleType

Bases: StrEnum

Discriminator identifying which concrete Schedule variant a value represents.

CRON class-attribute instance-attribute

CRON = 'cron'

Crontab-driven schedule.

INTERVAL class-attribute instance-attribute

INTERVAL = 'interval'

Fixed or backoff-based interval schedule.

OR class-attribute instance-attribute

OR = 'or'

Composite schedule that fires when any of its child schedules fires.

Connections

Buffer

A FIFO queue of timestamped byte chunks backed by a single contiguous bytearray.

Buffer accumulates incoming bytes from a connection along with the timestamp at which each chunk was observed. Bytes are stored in one contiguous bytearray, while a parallel list of (end_position, timestamp) entries records the boundary and time of each pushed chunk. Adjacent chunks with the same timestamp are merged into a single entry.

Indexing into the buffer is byte-oriented and zero-based relative to the current contents, independent of how many bytes have been popped from the front.

chunks property

chunks: Iterator[VirtualChunk]

Iterate over the original push boundaries as VirtualChunk views.

Adjacent pushes that shared a timestamp are emitted as one merged chunk.

data property

data: bytes

All buffered bytes as an immutable bytes object, cached between mutations.

earliest_timestamp property

earliest_timestamp: datetime | None

Timestamp of the oldest entry, or the latest timestamp if the buffer is empty.

latest_timestamp property

latest_timestamp: datetime | None

Timestamp of the most recently pushed chunk, or None if no data was ever pushed.

size property

size: int

Total number of bytes currently held in the buffer.

chunk_at

chunk_at(index: int) -> VirtualChunk | None

Return the VirtualChunk covering the byte at index, or None if out of bounds.

clear

clear() -> None

Remove all data and reset internal indices, leaving the buffer empty.

drain

drain(
    splitter: Splitter | None = None,
    *,
    linearize: bool = False
) -> list[Chunk]

Split the buffer with splitter, pop the consumed bytes, and return the chunks.

Parameters:

Name	Type	Description	Default
splitter	Splitter | None	The Splitter to use, defaults to SplitByChunk.	None
linearize	bool	See split().	False

Returns:

Type	Description
list[Chunk]	The drained Chunk objects in order. Bytes after the last split are left in the
list[Chunk]	buffer untouched.

extend

extend(records: Iterable[ChunkInput]) -> None

Push each item in records into the buffer in order.

pop

pop(count: int) -> Chunk | None

Remove the first count bytes from the buffer.

Parameters:

Name	Type	Description	Default
count	int	Number of bytes to remove from the front of the buffer. Values clamp to the current buffer size.	required

Returns:

Type	Description
Chunk | None	A Chunk carrying the popped bytes and the timestamp of the entry that contained
Chunk | None	the last popped byte, or None if count <= 0 or the buffer is empty.

pop_to

pop_to(size: int, by: int = 1) -> Chunk | None

Pop bytes in multiples of by until the buffer fits within size bytes.

Parameters:

Name	Type	Description	Default
size	int	Target maximum buffer size in bytes.	required
by	int	Granularity at which bytes are dropped, the actual pop count is rounded up to the next multiple of by.	1

Returns:

Type	Description
Chunk | None	The Chunk returned by the underlying pop(), or None when no bytes need to be
Chunk | None	dropped.

push

push(
    data: ToBytes, time: datetime | timedelta | None = None
) -> None

push(data: ChunkInput) -> None

push(
    data: ChunkInput,
    time: datetime | timedelta | None = None,
) -> None

push(
    data: ToBytes | ChunkInput,
    time: datetime | timedelta | None = None,
) -> None

Append data to the buffer with the given timestamp.

If time is None, default to utc() when no prior data exists, otherwise reuse the latest timestamp. If time is a timedelta, treat it as an offset from the latest timestamp (or from utc() when the buffer is empty). When data is a Chunk or (data, time) tuple, the explicit time argument takes precedence over the embedded time when provided.

Adjacent pushes that share a timestamp are merged into the same internal entry to keep the index compact.

Raises:

Type	Description
ValueError	If the resolved timestamp is earlier than latest_timestamp.

split

split(
    splitter: Splitter | None = None,
    *,
    linearize: bool = False
) -> Iterator[VirtualChunk]

Yield VirtualChunk views of the buffer split according to splitter.

Parameters:

Name	Type	Description	Default
splitter	Splitter | None	The Splitter to use, defaults to SplitByChunk which preserves the original push boundaries.	None
linearize	bool	If true, force each yielded chunk's timestamp to be strictly greater than the previous chunk's timestamp by nudging duplicates forward by one microsecond.	False

Yields:

Type	Description
VirtualChunk	VirtualChunk instances covering successive non-overlapping byte ranges.

timestamp_at

timestamp_at(index: int) -> datetime | None

Return the timestamp of the chunk containing the byte at index, or None.

Chunk dataclass

A timestamped block of bytes received from or sent over a connection.

data instance-attribute

data: bytes

The raw byte payload.

timestamp instance-attribute

timestamp: datetime

The time at which this chunk was observed.

Connection

Bases: DataObject, Tasklet

A managed byte-oriented connection driven by a Source transport.

Connection handles the full lifecycle of a single connection, including establishing the transport, buffering and splitting incoming bytes into messages, sending outbound data, and automatically reconnecting on failure. It runs as a background Tasklet managed by the owning component's ComponentConnectionManager.

__system__ property writable

__system__: ComponentSystem

Return the owning ComponentSystem, creating a default one if unset.

buffer property

buffer: bytes

Return a snapshot of the current receive buffer contents as bytes.

buffer_drop class-attribute instance-attribute

buffer_drop: ByteSize = Field(
    default=DEFAULT_BUFFER_DROP, gt=0
)

Number of bytes to drop from the front of the buffer on overflow.

buffer_read_size class-attribute instance-attribute

buffer_read_size: ByteSize = Field(
    default=DEFAULT_BUFFER_READ_SIZE, gt=0
)

Maximum number of bytes to read from the source in a single call.

buffer_size class-attribute instance-attribute

buffer_size: ByteSize = Field(
    default=DEFAULT_BUFFER_SIZE, gt=0
)

Maximum buffer size before overflow handling kicks in.

connect_timeout class-attribute instance-attribute

connect_timeout: PositiveTimeDelta | None = None

Maximum time to wait for the source to connect before giving up.

connected property

connected: bool

Return True if the connection is currently in the CONNECTED state.

connectivity property

connectivity: Connectivity

Return the current Connectivity state of this connection.

description class-attribute instance-attribute

description: NonBlankStr | None = None

What the connection carries, defaulting to the field's docstring.

label class-attribute instance-attribute

label: NonBlankStr | None = None

Human-readable name shown in place of name, defaulting to the field's title.

messages property

messages: BoundMessageManager

Return a BoundMessageManager scoped to messages on this connection.

name class-attribute instance-attribute

name: Name | None = None

Optional name used to identify this connection within its component.

receive_timeout class-attribute instance-attribute

receive_timeout: PositiveTimeDelta | None = None

Maximum time to wait for incoming data before emitting a timeout event.

reconnect_schedule class-attribute instance-attribute

reconnect_schedule: Schedule | None = field(
    default_factory=lambda: IntervalSchedule(
        interval=timedelta(seconds=1),
        multiplier=2,
        max=timedelta(seconds=60),
    )
)

Schedule governing delay between reconnection attempts after a failure.

source instance-attribute

source: Loaded[Source]

Transport endpoint that provides the underlying byte stream.

splitter class-attribute instance-attribute

splitter: Loaded[Splitter] | None = None

Strategy for splitting the receive buffer into discrete messages.

suffix class-attribute instance-attribute

suffix: MessageData | None = None

Bytes appended to outbound data when suffixed=True in send().

uri property

uri: str

Return the URI the underlying Source reads from.

__run__ async

__run__() -> None

Run the connection lifecycle loop.

Repeatedly attempt to connect using the configured reconnect schedule, then read incoming data, split it into messages, and emit events until the connection is lost or the task is cancelled.

__stop__ async

__stop__() -> None

Disconnect the connection when the tasklet is stopped.

connect async

connect() -> bool

Attempt to establish the connection via the configured Source.

Emit ConnectingEvent before the attempt, then ConnectedEvent or ConnectFailedEvent depending on the outcome. If connect_timeout is set and exceeded, emit ConnectTimeoutEvent and treat the attempt as failed.

Returns:

Type	Description
bool	True if the connection is established when this call returns.

disconnect async

disconnect() -> None

Gracefully close the connection and clear the receive buffer.

Emit DisconnectingEvent before teardown and DisconnectedEvent after. No-op if the connection is already disconnected.

receive async

receive(
    *,
    where: Callable[[Message], bool] | None = None,
    timeout: float | timedelta | None = None,
    default: T | Callable[[], T],
    **kwargs: Unpack[MessageFilterArgs]
) -> Message | T

receive(
    *,
    where: Callable[[Message], bool] | None = None,
    timeout: float | timedelta | None = None,
    default: ... = ...,
    **kwargs: Unpack[MessageFilterArgs]
) -> Message

receive(
    *,
    where: Callable[[Message], bool] | None = None,
    timeout: float | timedelta | None = None,
    default: T | Callable[[], T] = ...,
    **kwargs: Unpack[MessageFilterArgs]
) -> Message | T

Wait for and return the next matching message on this connection.

Messages are matched against the optional where predicate and any MessageFilter keyword arguments. If timeout expires before a match is found, return default (calling it first if it is callable) or raise TimeoutError when no default was provided.

Parameters:

Name	Type	Description	Default
where	Callable[[Message], bool] | None	Optional predicate that must return True for a message to match.	None
timeout	float | timedelta | None	Maximum time to wait, as seconds or a timedelta.	None
default	T | Callable[[], T]	Value (or zero-argument callable returning a value) to return on timeout. When omitted, TimeoutError is raised instead.	...
**kwargs	Unpack[MessageFilterArgs]	Additional MessageFilterArgs forwarded to MessageFilter.	{}

Returns:

Type	Description
Message | T	The first matching Message, or default if the timeout elapses.

Raises:

Type	Description
TimeoutError	If the timeout elapses and no default was supplied.

send async

send(data: ToBytes, suffixed: bool = True) -> Message

Send raw bytes through the connection, returning the sent Message.

There is no guarantee the returned message is actually received on the remote end, only that the data was transmitted to the transport.

Parameters:

Name	Type	Description	Default
data	ToBytes	Payload to send, converted to bytes before transmission.	required
suffixed	bool	When True, append self.suffix to the data if it is set and the data does not already end with the suffix.	True

Returns:

Type	Description
Message	The Message record created for the sent data.

Raises:

Type	Description
ConnectionInactive	If the connection is not currently connected.
ConnectionLost	If the underlying transport fails during the send.

ConnectionException

Bases: Exception

Base exception for connection-level errors.

ConnectionField

Bases: BoundField['Connection']

Pydantic field descriptor that declares a Connection slot on a component.

Connection-specific keyword arguments (e.g. source, splitter, buffer_size) are extracted from kwargs and folded into the defaults mapping so the connection is preconfigured when the component is instantiated.

Marker dataclass

Bases: Marker

Annotation marker identifying a field as a ConnectionField.

__init__

__init__(
    default: Any = ...,
    **kwargs: Unpack[ConnectionFieldArgs]
)

Create a ConnectionField with optional connection defaults.

Parameters:

Name	Type	Description	Default
default	Any	Default value for the field, or ... to require explicit assignment.	...
**kwargs	Unpack[ConnectionFieldArgs]	Connection-level keyword arguments and standard BoundFieldArgs. Any key that matches a Connection model field is moved into the defaults mapping automatically.	{}

ConnectionInactive

Bases: ConnectionException

Raised when an operation requires an active connection but the connection is not connected.

ConnectionLost

Bases: ConnectionException

Raised when an established connection is lost during a send or receive operation.

SplitByChunk

Bases: Splitter

Split on the original push boundaries recorded by the buffer.

split

split(buffer: Buffer) -> Iterator[int]

Yield the end index of each chunk stored in the buffer.

SplitByDelay

Bases: Splitter

Split whenever the gap between consecutive chunks meets or exceeds delay.

delay class-attribute instance-attribute

delay: PositiveTimeDelta = field(kw_only=False)

Minimum inter-chunk delay that triggers a split.

split

split(buffer: Buffer) -> Iterator[int]

Yield a boundary at each chunk whose gap from the prior chunk meets delay.

SplitByLine

Bases: Splitter

Split on each newline (\n), inclusive of the newline byte itself.

pattern property

pattern: Pattern[bytes]

The compiled regex used to locate line breaks.

split

split(buffer: Buffer) -> Iterator[int]

Yield the end index of each newline match found in the buffer.

SplitByRegex

Bases: Splitter

Split wherever a regular expression matches, with placement controlled by mode.

mode class-attribute instance-attribute

mode: Mode = 'suffix'

Where the match falls relative to its surrounding frames, see SplitByRegexMode.

pattern class-attribute instance-attribute

pattern: Pattern[bytes] = field(kw_only=False)

The compiled regex applied to the buffer's bytes.

split

split(buffer: Buffer) -> Iterator[int]

Yield boundary indices for each regex match, placed according to mode.

Splitter

Bases: Frozen

Determine where to slice a Buffer into framed pieces.

Subclasses yield exclusive end indices into the buffer's current contents. Each yielded index marks the boundary between two consecutive logical frames, the bytes from the previous yield (or zero) up to (but not including) the yielded index form one frame.

split abstractmethod

split(buffer: Buffer) -> Iterator[int]

Yield exclusive end indices marking frame boundaries within buffer.

Parameters:

Name	Type	Description	Default
buffer	Buffer	The Buffer to inspect. The splitter must not modify it.	required

Yields:

Type	Description
int	Strictly increasing exclusive end indices into buffer.

TCPSource

Bases: AnyIOSource

A Source that connects to a remote TCP host and port.

host instance-attribute

host: NonBlankStr

Hostname or IP address to connect to.

port instance-attribute

port: NonNegativeInt

TCP port to connect to.

uri property

uri: str

Return the tcp://host:port this source reads from.

UNIXSocketSource

Bases: AnyIOSource

A Source that connects to a UNIX domain socket. Unsupported on non-UNIX platforms.

socket instance-attribute

socket: NonBlankStr

Filesystem path of the UNIX domain socket to connect to.

uri property

uri: str

Return the unix://path this source reads from.

Unsplit

Bases: Splitter

Treat the entire buffer as a single frame.

split

split(buffer: Buffer) -> Iterator[int]

Yield the buffer's total size as a single boundary if the buffer is non-empty.

Connectivity

Bases: StrEnum

Represent the current connection state of a Connection.

A connection moves through these states as it attempts to establish or maintain its underlying transport, the engine surfaces this state to interested observers.

CONNECTED class-attribute instance-attribute

CONNECTED = 'connected'

The connection has an active transport ready to send and receive data.

CONNECTING class-attribute instance-attribute

CONNECTING = 'connecting'

The connection is actively attempting to establish its transport.

DISCONNECTED class-attribute instance-attribute

DISCONNECTED = 'disconnected'

The connection has no active transport and is not currently attempting to open one.

Client

Bases: Protocol

Protocol describing a connected peer that a Server handler can talk to.

receive async

receive(size: int = ...) -> bytes

Receive up to size bytes from the peer.

send async

send(data: bytes) -> None

Send data to the peer.

send_eof async

send_eof() -> None

Signal end-of-file to the peer. Should be idempotent.

Server

Bases: Component

Abstract accepting server that runs a handler per connected client.

Concrete subclasses implement serve() to accept connections and handle() to process each client. Implementations in this module bind the lifecycle to an anyio listener and wire in connection/disconnection events plus automatic rebind on failure.

auto_eof class-attribute instance-attribute

auto_eof: bool = True

If true, send EOF to the client after handle() returns when none was sent already.

rebind_on class-attribute instance-attribute

rebind_on: Schedule = IntervalSchedule(
    interval=timedelta(seconds=1),
    multiplier=2,
    max=timedelta(seconds=60),
)

Schedule that controls how long to wait before rebinding after a bind failure.

handle abstractmethod async

handle(client: ClientT) -> None

Process a single connected client. Called once per accepted connection.

serve abstractmethod async

serve() -> None

Run the server's accept loop until cancelled or the rebind schedule is exhausted.

TCPClient

Bases: _AnyIOClient[SocketStream]

Client bound to a TCP peer, exposing the peer's host and port.

host property

host: str

Remote peer's host or IP address.

port property

port: int

Remote peer's TCP port.

TCPServer

Bases: _AnyIOServer[TCPClient]

Server that binds a TCP listener to host:port and serves TCPClients.

host class-attribute instance-attribute

host: NonEmptyStr = '0.0.0.0'

Local interface to bind the listener on.

port instance-attribute

port: NonNegativeInt

Local TCP port to bind the listener on.

UNIXSocketClient

Bases: _AnyIOClient[SocketStream]

Client bound to a UNIX domain socket peer.

socket property

socket: str

Filesystem path of the remote peer's socket.

UNIXSocketServer

Bases: _AnyIOServer[UNIXSocketClient]

Server that binds a UNIX domain socket listener and serves UNIXSocketClients.

Unsupported on non-UNIX platforms, _validate_os rejects construction on other systems.

socket instance-attribute

socket: Path

Filesystem path at which to create the UNIX domain socket.

socket_mode class-attribute instance-attribute

socket_mode: NonNegativeInt | None = None

Optional file mode for the created socket. None keeps the platform default.

Channel

Bases: OutputChannel[T]

Pub/sub primitive that fans values out to readers and derived OutputChannels.

A Channel is the writable side, owners call put to publish a value and any attached readers or chained outputs receive it. Use output to expose a read-only handle to consumers without leaking the ability to publish.

output

output() -> OutputChannel[T]

Return a read-only OutputChannel view of this channel.

put

put(value: T) -> None

Publish value to every attached reader and chained output.

ChannelReader

Bases: AsyncIterator[T]

Async iterator and queue that buffers values delivered from an OutputChannel.

A ChannelReader registers itself with its source on construction, values published while the reader is attached are queued for retrieval via get or async iteration. The reader can be detached and reattached, and supports use as a context manager that detaches on exit.

Each reader is bound to the event loop that was running when it was constructed (or the first loop encountered if none was running at construction time). Values produced from other loops are delivered safely via call_soon_threadsafe, and get calls issued from a different loop are routed back to the bound loop.

attached property

attached: bool

Return whether this reader is currently registered with its source.

loop property

loop: AbstractEventLoop | None

Return the event loop this reader is bound to, if any.

source property

source: OutputChannel[T]

Return the output channel this reader is attached to.

__init__

__init__(source: OutputChannel[T]) -> None

Construct a reader bound to source and attach it.

Parameters:

Name	Type	Description	Default
source	OutputChannel[T]	The output channel to receive values from.	required

attach

attach() -> None

Register this reader with its source so it begins receiving values.

clear

clear() -> list[T]

Drain and return all currently queued values without waiting for new ones.

detach

detach() -> None

Unregister this reader from its source so it stops receiving values.

get async

get() -> T

Wait for and return the next value from the channel.

Returns:

Type	Description
T	The next value published to the source channel after this reader attached.

Raises:

Type	Description
RuntimeError	If no event loop has been bound to this reader yet and one is not currently running.

join async

join() -> None

Wait until all queued values have been consumed and acknowledged.

OutputChannel

Bases: AsyncIterable[T]

Read-only view of a Channel that fans values out to readers and derived outputs.

OutputChannel is the public side of the channel pub/sub primitive, callers can iterate it asynchronously, attach a ChannelReader directly, or build a derived output that filters or transforms incoming values. Only the originating Channel can publish, derived outputs forward values they receive from their source.

Readers and downstream outputs are tracked with weak references so that consumers are released as soon as they go out of scope, this avoids leaking resources when a long-lived component or engine produces values to short-lived listeners.

readers property

readers: Sequence[ChannelReader[T]]

Return a snapshot of currently attached readers.

__init__

__init__(source: OutputChannel[T] | None = None) -> None

Construct an output, optionally chained to a source output.

Parameters:

Name	Type	Description	Default
source	OutputChannel[T] | None	The upstream output to receive values from, or None for a root output owned directly by a Channel.	None

every

every(
    cls: type[O], /, *classes: type[O]
) -> OutputChannel[O]

Create a derived output that only forwards values matching the given types.

Parameters:

Name	Type	Description	Default
cls	type[O]	The first type to match against.	required
*classes	type[O]	Additional types to also match against, treated as a union.	()

Returns:

Type	Description
OutputChannel[O]	A new OutputChannel chained to this one, only isinstance matches pass through.

map

map(transform: Callable[[T], O]) -> OutputChannel[O]

Create a derived output that transforms each value via transform.

Parameters:

Name	Type	Description	Default
transform	Callable[[T], O]	Function applied to each value before forwarding.	required

Returns:

Type	Description
OutputChannel[O]	A new OutputChannel chained to this one carrying the transformed type.

read

read() -> ChannelReader[T]

Create and attach a new ChannelReader to this output.

where

where(where: Callable[[T], bool]) -> OutputChannel[T]

Create a derived output that only forwards values for which where returns truthy.

Parameters:

Name	Type	Description	Default
where	Callable[[T], bool]	Predicate evaluated for each value, exceptions are swallowed and treated as a non-match.	required

Returns:

Type	Description
OutputChannel[T]	A new OutputChannel chained to this one.

Particles

BinaryParticle

Bases: ParseableParticle[T]

Particle whose payload is parsed by unpack-ing fixed-layout binary fields.

The payload type (cls.Data) describes the binary layout via its field annotations, from_bytes invokes unpack to decode the bytes against that schema.

from_bytes classmethod

from_bytes(
    bytes: bytes,
    /,
    address: Address,
    timestamp: DateTime | None = None,
    span: tuple[int, int] | None = None,
) -> Self

Unpack a particle from fixed-layout binary bytes.

Parameters:

Name	Type	Description	Default
bytes	bytes	Raw bytes containing the binary payload.	required
address	Address	Address of the node or connection that produced the bytes.	required
timestamp	DateTime | None	Optional UTC timestamp, defaults to the current time.	None
span	tuple[int, int] | None	Optional (start, end) byte offsets. Defaults to the full extent of bytes.	None

Returns:

Type	Description
Self	A new instance of cls with data decoded from the binary layout.

Raises:

Type	Description
ParseFailed	If bytes cannot be unpacked into a valid payload.

BinaryRegexParticle

Bases: BinaryParticle[T], RegexParticle[T]

Hybrid particle that uses regex to find sync bytes and unpack to decode them.

The regex locates a candidate frame (typically using sync markers and a known length), then the matched byte range is handed to BinaryParticle.from_bytes for fixed-layout decoding via unpack.

from_match classmethod

from_match(
    match: Match[bytes],
    address: Address,
    timestamp: DateTime | None = None,
) -> Self

Decode a particle by unpacking the matched byte range as fixed-layout binary.

Delegates to BinaryParticle.from_bytes with the matched bytes and span.

Parameters:

Name	Type	Description	Default
match	Match[bytes]	A successful re.Match locating the binary frame.	required
address	Address	Address of the node or connection that produced the matched bytes.	required
timestamp	DateTime | None	Optional UTC timestamp, defaults to the current time.	None

Returns:

Type	Description
Self	A new instance of cls with data decoded from the matched bytes.

Raises:

Type	Description
ParseFailed	If the matched bytes cannot be unpacked into a valid payload.

GroupedRegexParticle

Bases: RegexParticle[T]

RegexParticle that maps named regex groups directly onto ParticleData fields.

Each capture group in the pattern is paired with a field of the bound ParticleData class, either positionally (in declaration order) or by name when the regex uses named groups. The captured byte values are validated through the data class to produce the final payload.

from_match classmethod

from_match(
    match: Match[bytes],
    address: Address,
    timestamp: DateTime | None = None,
) -> Self

Build a particle by mapping named regex groups onto ParticleData fields.

Each named group in match is paired with a field on cls.Data and validated through Pydantic to produce the payload.

Parameters:

Name	Type	Description	Default
match	Match[bytes]	A successful re.Match produced by cls.regex.	required
address	Address	Address of the node or connection that produced the matched bytes.	required
timestamp	DateTime | None	Optional UTC timestamp, defaults to the current time.	None

Returns:

Type	Description
Self	A new instance of cls with data populated from the capture groups.

Raises:

Type	Description
ParseFailed	If the captured group values fail validation against cls.Data.

group_indexes classmethod

group_indexes() -> Mapping[str, int]

Mapping from field name to the regex group index that captures it.

Positional groups are paired with payload fields in declaration order, then any explicitly named groups override or extend that mapping by group name.

group_names classmethod

group_names() -> Mapping[int, str]

Inverse of group_indexes, mapping each regex group index back to its field name.

ParseFailed dataclass

Bases: Exception

Raised when a ParseableParticle fails to decode its payload.

message instance-attribute

message: str = str(message)

Human-readable description of why parsing failed.

validation instance-attribute

validation: ValidationError | None = validation

Optional underlying Pydantic ValidationError when failure was due to validation.

ParseableParticle

Bases: Particle[DataT]

Abstract Particle that knows how to parse itself from raw bytes.

Concrete subclasses (BinaryParticle, RegexParticle, etc.) implement from_bytes to decode their payload from a byte sequence.

from_bytes abstractmethod classmethod

from_bytes(
    bytes: bytes,
    /,
    address: Address,
    timestamp: DateTime | None = None,
    span: tuple[int, int] | None = None,
) -> Self

Parse a particle instance from bytes.

Parameters:

Name	Type	Description	Default
bytes	bytes	Raw bytes containing a single particle's serialized form.	required
address	Address	Address of the node or connection that produced the bytes.	required
timestamp	DateTime | None	Optional UTC timestamp associated with the source bytes, defaults to the current time when not provided.	None
span	tuple[int, int] | None	Optional (start, end) byte offsets locating the particle within a larger buffer. Defaults to the full extent of bytes.	None

Returns:

Type	Description
Self	A new instance of cls populated from the parsed payload.

Raises:

Type	Description
ParseFailed	If bytes cannot be decoded into a valid payload.

from_message classmethod

from_message(message: Message) -> Self

Parse a particle from a Message, propagating its address and timestamp.

Parameters:

Name	Type	Description	Default
message	Message	The source message whose data bytes should be parsed.	required

Returns:

Type	Description
Self	A new instance of cls parsed from message.data.

Raises:

Type	Description
ParseFailed	If message.data cannot be decoded.

Particle

Bases: BaseRecord, ParticleCreate, ConcreteEntity, Generic[DataT]

A typed parsed record extracted from raw connection bytes.

A Particle carries the structured payload (data) parsed from one or more Message chunks together with metadata identifying when, where, and from which byte range it was extracted. Subclasses bind a concrete ParticleData type via the data annotation and set a stable type discriminator string.

data instance-attribute

data: DataT

Parsed payload, either a typed ParticleData subclass or a raw JSON object.

span class-attribute instance-attribute

span: Annotated[
    tuple[int, int] | None,
    Field(exclude_if=lambda value: value is None),
] = field(default=None, compare=False, hash=False)

Optional (start, end) byte offsets locating this particle in its source bytes.

Data classmethod

Data() -> type[DataT]

Resolved ParticleData subclass bound to the data field of cls.

convert

convert(cls: type[T]) -> T

Return a copy of this particle with data validated against cls.Data.

Parameters:

Name	Type	Description	Default
cls	type[T]	Concrete Particle subclass to convert to.	required

Returns:

Type	Description
T	A new cls instance carrying the same id, address, timestamp, type, and span
T	as self, with data re-validated against the target payload type.

Raises:

Type	Description
ValidationError	If self.data cannot be coerced to the target cls.Data type.

to_dynamic

to_dynamic() -> Particle

Return a base Particle whose data is a plain JSON object.

Useful when serializing or comparing particles without depending on the concrete payload class being importable.

Returns:

Type	Description
Particle	A new dynamic Particle carrying the same metadata with data reduced to a
Particle	plain JSON-serializable dictionary.

ParticleData

Bases: DataObject, Mapping[str, Any]

Base class for the typed payload of a Particle.

Subclasses define the parsed fields extracted from raw bytes. ParticleData exposes a Mapping interface over its attributes so payload values can be accessed both as attributes and as dictionary entries. Unknown fields are silently dropped during validation, allowing payloads to evolve without breaking older readers.

items

items() -> ItemsView[str, Any]

Return a view of the payload's (field_name, value) pairs.

keys

keys() -> KeysView[str]

Return a view of the payload's field names.

values

values() -> ValuesView[Any]

Return a view of the payload's field values.

RegexParticle

Bases: ParseableParticle[T]

Particle parsed by matching a regex pattern against raw bytes.

Subclasses set the regex class attribute to a compiled byte pattern and implement from_match to translate a re.Match into a particle instance.

regex class-attribute

regex: Pattern[bytes]

Compiled regex pattern used to locate and capture this particle's bytes.

extract classmethod

extract(
    data: Buffer,
    /,
    address: Address,
    timestamp: DateTime | None = None,
    errors: (
        Literal["ignore", "raise"]
        | Callable[[ParseFailed], Any]
    ) = "ignore",
) -> Iterable[Self]

Scan data for matches and yield successfully parsed particles.

Parameters:

Name	Type	Description	Default
data	Buffer	Buffer of raw bytes to search.	required
address	Address	Address to attach to each parsed particle.	required
timestamp	DateTime | None	Fallback timestamp used when the buffer cannot resolve one.	None
errors	Literal['ignore', 'raise'] | Callable[[ParseFailed], Any]	How to handle ParseFailed exceptions. "ignore" skips the failed match, "raise" re-raises, a callable receives the exception and the loop continues.	'ignore'

Yields:

Type	Description
Iterable[Self]	Each successfully parsed particle from data.

Raises:

Type	Description
ParseFailed	When errors is "raise" and a match fails to parse.

from_bytes classmethod

from_bytes(
    bytes: bytes,
    /,
    address: Address,
    timestamp: DateTime | None = None,
    span: tuple[int, int] | None = None,
) -> Self

Match the class regex against bytes and parse the result.

Parameters:

Name	Type	Description	Default
bytes	bytes	Raw bytes to match the regex against.	required
address	Address	Address of the node or connection that produced the bytes.	required
timestamp	DateTime | None	Optional UTC timestamp, defaults to the current time.	None
span	tuple[int, int] | None	Ignored, the span is derived from the regex match.	None

Returns:

Type	Description
Self	A new instance of cls built from the regex match.

Raises:

Type	Description
ParseFailed	If bytes do not match the regex pattern.

from_match abstractmethod classmethod

from_match(
    match: Match[bytes],
    /,
    address: Address,
    timestamp: DateTime | None = None,
) -> Self

Build a particle instance from a regex match.

Parameters:

Name	Type	Description	Default
match	Match[bytes]	A successful re.Match produced by cls.regex.	required
address	Address	Address of the node or connection that produced the matched bytes.	required
timestamp	DateTime | None	Optional UTC timestamp for the matched data, defaults to now.	None

Returns:

Type	Description
Self	A new instance of cls populated from the captured groups.

Raises:

Type	Description
ParseFailed	If the captured groups cannot be validated into a payload.

scan classmethod

scan(
    data: Buffer,
    /,
    address: Address,
    timestamp: DateTime | None = None,
) -> Iterable[RegexParticleMatch[Self]]

Yield every regex match in data paired with its resolved chunk timestamp.

Parameters:

Name	Type	Description	Default
data	Buffer	Buffer of raw bytes to search.	required
address	Address	Address to attach to each yielded match.	required
timestamp	DateTime | None	Fallback timestamp used when the buffer cannot resolve a chunk timestamp for a match's location.	None

Yields:

Type	Description
Iterable[RegexParticleMatch[Self]]	RegexParticleMatch for every non-overlapping match, skipping any match for
Iterable[RegexParticleMatch[Self]]	which neither the buffer nor the fallback can supply a timestamp.

RecordType

Bases: StrEnum

Discriminator for the variants of Record.

RecordType is a strict subset of ItemType, every record type also exists as an item type and the two enums share their string values to keep cross-references unambiguous.

cls property

cls: type[Record]

Return the concrete Record subclass associated with this variant.

upcast

upcast() -> ItemType

Return the equivalent ItemType value, widening from record into item.

Addressing

Address

Bases: DynamicAddress

An absolute address pointing to a single component or the engine.

The reserved value is ~ (the engine). Every component address is @ followed by dot-separated names, e.g. @parent.child.grandchild.

ENGINE classmethod

ENGINE() -> Address

The engine address (~).

contains

contains(other: Address) -> bool

Return True if other is self or a descendant of self.

The engine address contains every other address.

engine classmethod

engine() -> Address

Return the engine address (~).

AddressSelector

A pattern that selects one or more component addresses.

Selectors support a small DSL for matching addresses by exact value or by structural relationship. A selector consists of one or more pipe-separated segments where each segment is either a literal address (e.g. @sensor), the special base ~ (engine), or a base followed by a modifier. The base @ selects every component and requires a modifier, it is not a matchable address on its own:

• :all matches the base itself plus every descendant.
• :children matches only the immediate children of the base.
• :descendants matches every descendant of the base, but not the base itself.

Instances are interned by their textual form so equal selectors share memory.

REGEX class-attribute instance-attribute

REGEX: Final = compile(
    f"^(?:{_SEGMENT})(\|(?:{_SEGMENT}))*$"
)

Regex used to validate the textual form of a selector.

segments property

segments: Sequence[AddressSelector]

The pipe-separated segments of the selector, each as its own AddressSelector.

text property

text: str

The selector's textual form.

as_absolute

as_absolute(root: Address | None) -> AddressSelector

Resolve relative selector segments against root to produce an absolute selector.

Segments that already begin with ~, @, or : are left untouched. Everything else is treated as relative to root and gets prefixed accordingly.

Parameters:

Name	Type	Description	Default
root	Address | None	Address that relative segments resolve against. None or the engine resolve against every component.	required

Returns:

Type	Description
AddressSelector	An equivalent selector whose segments are all absolute.

matches

matches(address: Address, root: Address | None) -> bool

Check whether address is selected by this selector when resolved against root.

Parameters:

Name	Type	Description	Default
address	Address	Component address to test.	required
root	Address | None	Reference address used to resolve any relative segments. None or the engine resolve against every component.	required

Returns:

Type	Description
bool	True if any segment matches address, False otherwise.

DynamicAddress

Bases: AddressSelector

An address that may be relative or absolute, with helpers for path manipulation.

Unlike AddressSelector, a DynamicAddress always refers to a single concrete location rather than a pattern. It exposes properties for navigating the address tree (parent, ancestors, path, etc.) and operators for joining segments (/).

ancestors property

ancestors: Sequence[Self]

All ancestor addresses, from immediate parent up to the top-level ancestor.

base property

base: str | None

The address text with any selector modifier (:all, etc.) stripped.

container property

container: Self

The parent address, or ~ (engine) if this address has no parent.

depth property

depth: int

Number of name segments in the address. ~ has depth 0.

is_absolute property

is_absolute: bool

True if this address starts at the engine or a component rather than being relative.

is_engine property

is_engine: bool

True if this address refers to the engine itself (~).

is_relative property

is_relative: bool

True if this address is not absolute and must be resolved against a base.

name property

name: str | None

The trailing name segment of the address, or None for ~.

names property

names: Sequence[str]

The address segments as bare names, without the leading @.

parent property

parent: Self | None

The parent address, or None if this address has no parent.

path property

path: Sequence[Self]

The address chain from the top-level ancestor to self, inclusive.

all

all() -> AddressSelector

Return a selector matching this address and every descendant.

as_absolute

as_absolute(root: Address | None) -> Address

Resolve this address against root and return the result as an absolute Address.

If this address is already absolute, return it directly. Otherwise, join it onto root.

Parameters:

Name	Type	Description	Default
root	Address | None	Base address to resolve against when this address is relative. None or the engine resolve against every component.	required

Returns:

Type	Description
Address	An absolute Address.

as_relative

as_relative() -> DynamicAddress | None

Strip the absolute prefix and return the relative form, or None if not possible.

children

children() -> AddressSelector

Return a selector matching only the immediate children of this address.

descendants

descendants() -> AddressSelector

Return a selector matching every descendant of this address.

modify

modify(
    modifier: Literal["all", "descendants", "children"]
) -> AddressSelector

Return a selector that matches self modified by modifier.

Ref module-attribute

Ref = Reference

Reference

Lazy handle to a Component that behaves like the component itself once resolved.

A Reference can be constructed from a concrete Component, another Reference, a DynamicAddress, or a string address. When the target is an address, resolution happens lazily against the reference's root component at access time. Attribute access, operators, and most dunder methods are proxied through to the resolved component so most call sites can treat a Reference as if it were the underlying component.

Use Reference[SomeComponent] to constrain the target type at construction time. The returned subclass enforces the constraint and forwards the proxy protocol for the constrained component's methods.

__reference_constraint__ class-attribute instance-attribute

__reference_constraint__: type[Component] | None = None

Component subclass the reference must resolve to, or None for no constraint.

__init__

__init__(
    target: Component | Reference | DynamicAddress | str,
    root: Component | Reference | None = None,
) -> None

Construct a reference pointing at target.

Parameters:

Name	Type	Description	Default
target	Component | Reference | DynamicAddress | str	Component, existing reference, dynamic address, or string address to resolve. Strings are parsed as DynamicAddress values.	required
root	Component | Reference | None	Component to resolve address-based targets against. Required when the target is a DynamicAddress or string address and the reference will be dereferenced.	None

Raises:

Type	Description
ValueError	If target is not a supported type, or if it is a component whose type does not satisfy __reference_constraint__.

validate classmethod

validate(value: Any) -> Self | None

Coerce value into an instance of this reference type.

Parameters:

Name	Type	Description	Default
value	Any	A Reference, Component, DynamicAddress, string address, or a structured value that validates as the constraint type.	required

Returns:

Type	Description
Self | None	A reference wrapping the coerced target, or None when value is None.

unref

unref(component: T) -> T

unref(component: T | None) -> T | None

unref(component: MaybeReference | None) -> Component | None

unref(component: MaybeReference | None) -> Component | None

Return the concrete component behind a value, resolving it if it is a reference.

Parameters:

Name	Type	Description	Default
component	MaybeReference | None	A component, a reference, or None.	required

Returns:

Type	Description
Component | None	The underlying component, or None if the input is None or an unresolvable
Component | None	reference.

Directory

Bases: PathLike[str]

A filesystem directory handle with helpers for reading, writing, and traversing contents.

Directory resolves child paths relative to its base path and exposes helpers for common filesystem operations. Directories can be marked temporary, in which case they are removed when the handle is garbage collected.

path property

path: Path

Absolute Path to the directory.

temporary property

temporary: bool

Whether the directory is removed when this handle is garbage collected.

__init__

__init__(
    path: StrPath | None = None,
    parent: Self | None = None,
    temporary: bool | None = None,
) -> None

Construct a directory handle rooted at path.

Parameters:

Name	Type	Description	Default
path	StrPath | None	Absolute or relative path to the directory. When parent is provided, path is resolved relative to it. When None, a unique temporary path is generated under the system temp directory.	None
parent	Self | None	Optional parent Directory, used for resolving path and recorded for reference.	None
temporary	bool | None	When True, remove the directory in __del__. Defaults to True when path is None, and False otherwise.	None

create

create(
    *, mkdirs: bool = True, exist_ok: bool = True
) -> None

Create this directory on disk.

Parameters:

Name	Type	Description	Default
mkdirs	bool	When True, create intermediate parent directories as needed.	True
exist_ok	bool	When True, silently succeed if the directory already exists.	True

exists

exists(path: StrPath | None = None) -> bool

Return whether the given path exists.

Parameters:

Name	Type	Description	Default
path	StrPath | None	Path relative to this directory, an absolute path, or None to check whether this directory itself exists.	None

Returns:

Type	Description
bool	True if the path exists. When checking this directory itself, the path must
bool	be a directory, not a file.

iter_subdirs

iter_subdirs(path: StrPath | None = None) -> Iterable[Self]

Yield child Directory handles for each subdirectory directly inside path.

Parameters:

Name	Type	Description	Default
path	StrPath | None	Path relative to this directory, an absolute path, or None to list this directory itself.	None

Yields:

Type	Description
Iterable[Self]	Directory handles for each direct child that is a directory.

iter_subpaths

iter_subpaths(
    path: StrPath | None = None,
) -> Iterable[Path]

Yield absolute paths for each entry directly inside path.

Parameters:

Name	Type	Description	Default
path	StrPath | None	Path relative to this directory, an absolute path, or None to list this directory itself.	None

Yields:

Type	Description
Iterable[Path]	Absolute paths for each direct child entry, in filesystem order.

move

move(
    source: StrPath,
    destination: StrPath,
    *,
    mkdirs: bool | None = None
) -> None

Move a file or directory from source to destination.

Parameters:

Name	Type	Description	Default
source	StrPath	Source path relative to this directory, or an absolute path.	required
destination	StrPath	Destination path relative to this directory, or an absolute path.	required
mkdirs	bool | None	When True, ensure the destination's parent directory exists. Defaults to True.	None

open

open(
    path: StrPath,
    mode: OpenTextMode = "r",
    buffering: int = ...,
    encoding: str | None = None,
    errors: str | None = None,
    newline: str | None = None,
    closefd: bool = True,
    *,
    mkdirs: bool | None = None,
    **kwargs: Any
) -> IO[str]

open(
    path: StrPath,
    mode: OpenBinaryMode,
    buffering: int = ...,
    encoding: str | None = None,
    errors: str | None = None,
    newline: str | None = None,
    closefd: bool = True,
    *,
    mkdirs: bool | None = None,
    **kwargs: Any
) -> IO[bytes]

open(
    path: StrPath,
    mode: OpenMode = "r",
    buffering: int = -1,
    encoding: str | None = None,
    errors: str | None = None,
    newline: str | None = None,
    closefd: bool = True,
    *,
    mkdirs: bool | None = None,
    **kwargs: Any
) -> IO[str] | IO[bytes]

Open a file inside the directory, creating parent directories on demand.

Parameters:

Name	Type	Description	Default
path	StrPath	Path relative to this directory, or an absolute path.	required
mode	OpenMode	Open mode string passed to open.	'r'
buffering	int	Buffering policy passed to open.	-1
encoding	str | None	Text encoding passed to open.	None
errors	str | None	Error handler name passed to open.	None
newline	str | None	Newline handling passed to open.	None
closefd	bool	Whether open should close the underlying file descriptor.	True
mkdirs	bool | None	When True, ensure the parent directory exists. Defaults to True when the mode is write or append, False otherwise.	None
**kwargs	Any	Additional keyword arguments forwarded to open.	{}

Returns:

Type	Description
IO[str] | IO[bytes]	The opened file object, matching the mode's text or binary variant.

remove

remove(
    path: StrPath | None = None, *, recursive: bool = True
) -> None

Remove a file or directory.

Parameters:

Name	Type	Description	Default
path	StrPath | None	Path relative to this directory, an absolute path, or None to remove this directory itself.	None
recursive	bool	When removing a directory, delete its contents too. Has no effect when removing a file.	True

subdir

subdir(
    path: StrPath, *, temporary: bool | None = None
) -> Self

Return a child Directory rooted at path relative to this directory.

Parameters:

Name	Type	Description	Default
path	StrPath	Relative path to the child directory.	required
temporary	bool | None	Forwarded to the child Directory, see Directory.__init__.	None

Returns:

Type	Description
Self	A new Directory whose parent is this one.

subdirs

subdirs(path: StrPath | None = None) -> list[Self]

Return the list of direct child directories under path, see iter_subdirs.

subpath

subpath(path: StrPath) -> Path

Return the absolute Path obtained by joining path onto this directory.

subpaths

subpaths(path: StrPath | None = None) -> list[Path]

Return the list of direct child paths under path, see iter_subpaths.

touch

touch(path: StrPath) -> None

Ensure the file at path exists, creating an empty file if needed.

Parameters:

Name	Type	Description	Default
path	StrPath	Path relative to this directory, or an absolute path.	required

Data

DataObject

Bases: _DataObjectProtocols

Base class for validated, serializable data records in Ceres.

DataObject combines Python's dataclasses with Pydantic's validation and serialization. Subclasses define typed fields as dataclass attributes, gain automatic validation on construction, JSON serialization, and fields-set tracking (knowing which fields were explicitly provided versus defaulted).

Create subclasses by inheriting from DataObject and declaring fields as normal dataclass attributes. Use slots=True in the class keyword arguments for concrete classes to allocate per-field slots. Use abstract=True for base classes that should not be instantiated directly. Pass frozen=True (or inherit from DataObject.Frozen) for immutable instances.

Instances can be constructed without validation using construct() or create() for pre-validated data, or with full Pydantic validation by calling the class directly.

__data_object_fields_set__ class-attribute instance-attribute

__data_object_fields_set__: FieldsSet = field(
    init=False, repr=False, compare=False
)

Set of field names explicitly set during initialization or assigned afterward.

Provide equivalent set/unset functionality to Pydantic's BaseModel.

Unit

Unit(unit: str) -> Any

Annotated metadata recording the unit a field's values are measured in.

The unit is published as unit in the field's JSON schema, where interfaces such as the console read it. For example, Annotated[float, Unit("degC")].

EntityType

Bases: StrEnum

Discriminator naming each variant of Entity.

The string values double as the canonical entity name in URLs, configuration files, and stored records. Aliases such as plural forms are accepted via __new__.

cls property

cls: type[Entity]

Return the concrete entity class associated with this variant.

Returns:

Type	Description
type[Entity]	The Entity subclass matching this enum value.

Raises:

Type	Description
ValueError	If the enum value has no associated class, this should be unreachable for valid EntityType members.

from_class classmethod

from_class(source: type[Entity]) -> EntityType

Return the EntityType matching an entity class.

Parameters:

Name	Type	Description	Default
source	type[Entity]	An entity class such as Message or Particle.	required

Returns:

Type	Description
EntityType	The EntityType member that corresponds to source.

Raises:

Type	Description
ValueError	If source is not one of the recognized entity classes.

ItemType

Bases: StrEnum

Discriminator identifying which concrete record type an Item is.

ItemType is a strict subset of EntityType covering only the user-facing record kinds, it upcasts to EntityType when interaction with the broader entity system is required.

cls property

cls: type[Item]

Return the concrete record class associated with this item type.

upcast

upcast() -> EntityType

Return the matching EntityType value for this item type.

Variable

Bases: BaseAddressEntity, VariableCreate, ConcreteEntity

Named JSON-serializable value owned by an addressed node.

Variables store configuration and state keyed by (address, name). Internal variables use the __name__ convention and are treated specially by some components. Assignments are broadcast as VariableAssignedEvent so listeners see updates in real time.

Setting

Bases: SettingCreate, ConcreteEntity

Per-user named value used to persist user preferences and application state.

Settings are keyed by (user_id, name) and store arbitrary JSON-serializable values.

Statistics

Bases: DataObject

Aggregated counts rolled up for a single address across its subtree of descendants.

address instance-attribute

address: Address

Address these statistics describe.

alerts class-attribute instance-attribute

alerts: AlertStatistics = Field(
    default_factory=AlertStatistics
)

Alert counts aggregated for this address and its descendants.

Events

Event

Bases: DataObject

Base class for all events emitted through the system.

Events propagate up through containing components and out to listeners. Each event carries the address of its originating component or engine, a timestamp, a stable string type discriminator, and a severity level.

address class-attribute instance-attribute

address: Address = cast('Address', None)

Address of the component or engine that emitted the event.

id class-attribute instance-attribute

id: UUID = Field(default_factory=uuid7)

Unique identifier for the event, defaulting to a time-ordered UUID7.

level class-attribute instance-attribute

level: Level = INFO

Severity level used for logging and filtering.

timestamp class-attribute instance-attribute

timestamp: DateTime = Field(default_factory=utc)

When the event occurred, in UTC.

type instance-attribute

type: str

Discriminator string identifying the concrete event class.

Message

Bases: BaseRecord, MessageCreate, ConcreteEntity

A chunk of data sent or received on a connection.

Each Message captures the raw bytes that flowed across a connection along with metadata identifying which connection produced it and which direction it traveled. Messages are the raw input from which Particle instances are typically parsed.

MessageDirection

Bases: StrEnum

Direction of a message on a connection, either outbound or inbound.

Alert

Bases: BaseRecord, AlertCreate, ConcreteEntity

Severity-tagged event raised by a component or engine and persisted as a record.

Alerts are how components and the engine surface notable conditions to operators, ranging from routine status updates to critical failures. Each alert carries a severity level, a discriminator type, and optional structured data.

LogEntry

Bases: BaseRecord, LogEntryCreate, ConcreteEntity

Text-content log line emitted by a component or engine and persisted as a record.

Log entries are the primary human-readable trail of activity in the system. Each entry carries a severity level and rendered content, and is produced via BoundLogManager on a running node.

Notification

Bases: DataObject

Message payload delivered by a Notifier to one or more recipients.

content class-attribute instance-attribute

content: str | None = None

Body of the notification, or None for a subject-only message.

content_type class-attribute instance-attribute

content_type: NonBlankStr = 'text/plain'

MIME type describing how to interpret content.

subject instance-attribute

subject: NonBlankStr

Short summary line shown as the notification's subject.

Notifier

Bases: Component

Component that delivers Notification payloads to external recipients.

Subclasses implement notify() to dispatch through a specific transport such as SMTP, SMS, or a chat service.

notify abstractmethod async

notify(
    notification: Notification,
    recipients: Iterable[NonBlankStr],
) -> None

Deliver a notification to the given recipients.

Parameters:

Name	Type	Description	Default
notification	Notification	Notification payload to deliver.	required
recipients	Iterable[NonBlankStr]	Addresses to deliver the notification to.	required

SMTPNotifier

Bases: Notifier

Notifier that delivers notifications as email messages over SMTP.

host instance-attribute

host: NonBlankStr

Hostname of the SMTP server to connect to.

password class-attribute instance-attribute

password: SecretStr | None = Field(None, min_length=1)

Optional password for SMTP authentication.

port instance-attribute

port: NonNegativeInt

TCP port of the SMTP server.

sender instance-attribute

sender: NonBlankStr

Address used as the From header on outgoing messages.

use_starttls class-attribute instance-attribute

use_starttls: bool = False

Upgrade the SMTP connection to TLS via STARTTLS after the initial handshake.

use_tls class-attribute instance-attribute

use_tls: bool = False

Connect with TLS immediately when opening the SMTP connection.

username class-attribute instance-attribute

username: NonBlankStr | None = None

Optional username for SMTP authentication.

notify async

notify(
    notification: Notification,
    recipients: Iterable[NonBlankStr],
) -> None

Send notification as an email to recipients over the configured SMTP server.

Parameters:

Name	Type	Description	Default
notification	Notification	Notification payload to deliver.	required
recipients	Iterable[NonBlankStr]	Email addresses to send the message to.	required

Dispatch

Bases: DataObject

Configuration for a single notification dispatch driven by alert activity.

A dispatch describes which alerts to summarize, who to notify, and optionally when to send the notification. The owning Dispatcher component runs each dispatch on its schedule, gathers matching alerts, renders a notification, and delivers it through the configured Notifier.

alerts instance-attribute

alerts: AlertFilter

Filter selecting which alerts contribute to this dispatch.

description class-attribute instance-attribute

description: NonBlankStr | None = None

Optional human-readable description rendered in the notification body.

recipients instance-attribute

recipients: list[str]

Recipient identifiers passed through to the configured Notifier.

schedule class-attribute instance-attribute

schedule: ScheduleExpr | None = None

Schedule controlling automatic dispatching, omit to dispatch only on demand.

signature class-attribute instance-attribute

signature: NonBlankStr | None = None

Optional signature appended to the notification body.

subject instance-attribute

subject: NonBlankStr

Subject line used as both the notification title and the dispatch identifier.

DispatchWriter

Render a Dispatch and a list of matching alerts into a Notification.

Implementations format alerts into the notification body and content type best suited to the delivery channel (for example HTML email or plaintext SMS).

write abstractmethod async

write(
    dispatch: Dispatch, alerts: Sequence[Alert]
) -> Notification

Render alerts for dispatch into a Notification.

Parameters:

Name	Type	Description	Default
dispatch	Dispatch	The dispatch configuration being rendered.	required
alerts	Sequence[Alert]	Alerts that matched the dispatch's filter, in arbitrary order.	required

Returns:

Type	Description
Notification	A Notification ready to hand to a Notifier for delivery.

Dispatcher

Bases: Component

Component that summarizes alerts and delivers notifications on a schedule.

The dispatcher owns a list of Dispatch configurations, registers a job for each one with a schedule, and renders matching alerts via a DispatchWriter before handing the result to a Notifier. Dispatches without a schedule can still be triggered on demand by calling the dispatch action directly.

dispatches class-attribute instance-attribute

dispatches: Sequence[Dispatch] = field(default_factory=list)

Dispatches managed by this component.

notifier instance-attribute

notifier: Ref[Notifier]

Reference to the Notifier component that delivers rendered notifications.

writer instance-attribute

writer: Loaded[DispatchWriter]

Writer responsible for rendering a Dispatch and its alerts into a Notification.

dispatch async

dispatch(dispatch: Dispatch) -> None

Run a dispatch end-to-end, gather alerts, render, then notify.

Parameters:

Name	Type	Description	Default
dispatch	Dispatch	The dispatch to execute.	required

routine__schedule_dispatches async

routine__schedule_dispatches() -> None

Register a scheduled job for each dispatch that defines a schedule.

HTMLDispatchWriter

Bases: DispatchWriter

DispatchWriter that renders alerts as an HTML email via a Jinja template.

Alerts are grouped first by severity level (highest first), then by the combination of source address, alert type, and serialized data, this collapses repeated alerts into a single row while preserving full timestamp history for each group.

write async

write(
    dispatch: Dispatch, alerts: Sequence[Alert]
) -> Notification

Render alerts into an HTML email notification using the built-in Jinja template.

Alerts are grouped by severity level, then collapsed by (address, type, data), and rendered with Markdown support for the dispatch description.

Parameters:

Name	Type	Description	Default
dispatch	Dispatch	The dispatch configuration being rendered.	required
alerts	Sequence[Alert]	Alerts that matched the dispatch's filter.	required

Returns:

Type	Description
Notification	A Notification with content_type set to text/html.

Level

Bases: OrderedStrEnum

Severity levels used throughout Ceres for events, alerts, and logs.

Levels are ordered from least to most severe, so comparisons like level >= Level.WARNING behave as expected. The string values match the lowercase names of the standard logging module levels.

CRITICAL class-attribute instance-attribute

CRITICAL = 'critical'

A very serious error, the program may be unable to continue running.

DEBUG class-attribute instance-attribute

DEBUG = 'debug'

Detailed diagnostic information, typically only enabled when troubleshooting.

ERROR class-attribute instance-attribute

ERROR = 'error'

A serious problem occurred, the operation failed.

INFO class-attribute instance-attribute

INFO = 'info'

Routine operational messages confirming things are working as expected.

WARNING class-attribute instance-attribute

WARNING = 'warning'

Something unexpected occurred, or a problem may occur in the near future.

from_int classmethod

from_int(value: int) -> Level

Convert an integer logging module level to a Level.

Parameters:

Name	Type	Description	Default
value	int	An integer matching one of the standard logging module level constants (e.g. logging.WARNING).	required

Returns:

Type	Description
Level	The Level member corresponding to the given integer.

Raises:

Type	Description
TypeError	If value is not an integer.
ValueError	If value does not match any known log level.

to_int

to_int() -> int

Convert this level to its corresponding integer value from the logging module.

Returns:

Type	Description
int	The integer constant from the standard logging module that corresponds to this
int	severity level.

Status

Bases: DataObject

Snapshot of a node's runtime status at a point in time.

address instance-attribute

address: Address

Address of the node this status describes.

connections class-attribute instance-attribute

connections: list[ConnectionStatus] = Field(
    default_factory=list
)

Per-connection connectivity states, empty when the component has no named connections.

connectivity class-attribute instance-attribute

connectivity: Connectivity | None = None

Overall connectivity when the component defines __connectivity__, otherwise None.

enabled class-attribute instance-attribute

enabled: bool | None = None

Whether the node is enabled, or None if the node has no enable/disable concept.

running instance-attribute

running: bool

Whether the node is currently running.

Sieve

Bases: DataObject

Extract Particle instances from a stream of Message objects.

Subclasses implement process() to consume an asynchronous stream of messages and yield parsed particles. A Sieve is the bridge between the raw Message data flowing from a Connection and the structured Particle objects that downstream consumers care about.

process abstractmethod

process(
    messages: AsyncIterable[Message],
) -> AsyncIterator[T]

Consume messages and yield extracted particles.

Parameters:

Name	Type	Description	Default
messages	AsyncIterable[Message]	An asynchronous stream of Message objects to parse.	required

Yields:

Type	Description
AsyncIterator[T]	Particles parsed from the message stream.

Access Control

User

Bases: BaseUUIDEntity, UserCreate, ConcreteEntity

Authenticated account with an admin flag that governs access to workspaces and resources.

Each user has a unique username, an email, a hashed password, and an admin flag that grants full access when set. Users may additionally be granted or restricted from specific workspaces and components via memberships and permission grants.

Group

Bases: BaseUUIDEntity, GroupCreate, ConcreteEntity

Named collection of users used to grant shared component permissions.

Each group has a unique name and an optional description. Users join a group through a GroupMembership, which is later used to grant component access levels to every member at once.

GroupMembership

Bases: GroupMembershipCreate, ConcreteEntity

Association record linking a User to a Group.

GroupPermission

Bases: GroupPermissionCreate, ConcreteEntity

Permission grant linking a Group to a component access level on a target.

PermissionTargetType

Bases: StrEnum

Discriminator for the kind of target a permission grant applies to.

ALL class-attribute instance-attribute

ALL = 'all'

Grant applies to every component. The target string is empty.

COMPONENT class-attribute instance-attribute

COMPONENT = 'component'

Grant applies to a specific component by address.

TAG class-attribute instance-attribute

TAG = 'tag'

Grant applies to all components carrying a given tag.

UserPermission

Bases: UserPermissionCreate, ConcreteEntity

Permission grant linking a User to a component access level on a target.

Workspace

Bases: BaseUUIDEntity, WorkspaceCreate, ConcreteEntity

Named arrangement of widgets, placed on a component or on the engine root.

A workspace has no access settings of its own. It inherits them from its placement, so whoever can view a component can view the workspaces placed on it, and whoever can manage that component can edit them. Setting owner_id makes a workspace private to one user, who is then the only person who can see or change it.

Database

Database

Asynchronous database handle backing all persisted Ceres state.

Database owns the native store every query runs through, exposes cached entity managers for every persisted record type, and handles one-time schema initialization. Instantiating the base class dispatches to the appropriate concrete subclass based on the configuration, so Database(SQLiteDatabaseConfig()) returns a SQLiteDatabase and Database(PostgresDatabaseConfig(...)) returns a PostgresDatabase. Omitting the config entirely dispatches on whatever default_database_config returns.

alerts cached property

alerts: AlertManager

Manager for Alert records.

config property

config: DatabaseConfig

Configuration object the database was constructed from.

ddl property

ddl: list[str]

Every script that creates this backend's schema, in the order they run.

See migration_ddl for the ordering and how dialect-specific scripts render.

group_memberships cached property

group_memberships: GroupMembershipManager

Manager for GroupMembership records.

group_permissions cached property

group_permissions: GroupPermissionManager

Manager for GroupPermission records.

groups cached property

groups: GroupManager

Manager for Group records.

id property

id: UUID

Identifier unique to this Database instance for this process.

logs cached property

logs: LogManager

Manager for log entry records.

messages cached property

messages: MessageManager

Manager for Message records.

particles cached property

particles: ParticleManager

Manager for Particle records.

settings cached property

settings: SettingManager

Manager for Setting records.

statistics cached property

statistics: StatisticsManager

Manager for aggregate statistics across persisted records.

type property

type: DatabaseType

Backend kind, one of DatabaseType.SQLITE, DatabaseType.TURSO, or DatabaseType.POSTGRES.

user_permissions cached property

user_permissions: UserPermissionManager

Manager for UserPermission records.

users cached property

users: UserManager

Manager for User records.

variables cached property

variables: VariableManager

Manager for Variable records.

workspace_edits cached property

workspace_edits: WorkspaceEditManager

Manager for WorkspaceEdit records.

workspaces cached property

workspaces: WorkspaceManager

Manager for Workspace records.

assert_schema_current async

assert_schema_current() -> None

Verify the database schema matches this version of ceres.

Raises:

Type	Description
DatabaseVersionError	If migrations are pending or unknown migrations are applied.

clear async

clear() -> None

Delete every row from every known entity table, preserving the schema itself.

The tables come from the native layer, which lists a table before anything that references it so deleting in reverse empties a table only once nothing points at it. Adding a table there puts it in the right place by the same rule.

dispose async

dispose() -> None

Let go of this database's pools so the connections they hold can close.

A close statement runs as each operation returns its connection rather than here, which makes it run at all on a backend that opens one connection per operation. What this releases is the pools themselves.

A disposed database is reusable. Reaching it again opens fresh connections and because the bootstrap flag goes with the pools, re-migrates. That last part matters for a temporary database, whose file is deleted on the way out and would otherwise be reopened empty and treated as migrated.

Waiting on the migration lock keeps a database from being disposed out from under its own bootstrap. A component stops as soon as it runs out of work, and the stop disposes its database so a query started from outside the component can be partway through ready when that happens. Closing the connections underneath it fails the migration, and the failure names neither the disposal nor the caller.

get_applied_migrations async

get_applied_migrations() -> list[Migration]

Return known migrations recorded as applied, in application order.

get_pending_migrations async

get_pending_migrations() -> list[Migration]

Return known migrations that have not been applied, in application order.

get_unknown_migrations async

get_unknown_migrations() -> list[int]

Return applied migration IDs this version of ceres does not know about.

hash_password async

hash_password(password: str) -> PasswordHash

Hash a plaintext password using the configured hashing parameters.

The hashing work runs on a worker thread so it does not block the event loop.

Parameters:

Name	Type	Description	Default
password	str	Plaintext password to hash.	required

Returns:

Type	Description
PasswordHash	The resulting password hash.

initialized async

initialized() -> bool

Check whether this schema has been created in the database.

The question is whether any table this schema owns is there, not whether the database holds a table at all. A configuration's init hook runs on connections opened before any migration has, and one that creates a table of its own would otherwise make an empty database look bootstrapped, leaving a project whose migrations never run and whose schema is never created.

Returns:

Type	Description
bool	True if any table this schema owns is present, False otherwise.

migrate async

migrate(
    reporter: MigrationReporter | None = None,
) -> list[int]

Apply every pending migration in order, recording each as it completes.

Holds an instance-level lock for the duration of the call so concurrent callers on the same Database instance apply migrations one at a time instead of racing to insert the same migration ID.

Parameters:

Name	Type	Description	Default
reporter	MigrationReporter | None	Told which migration is starting and when it finished, for a caller showing progress. A migration runs as one script so there is nothing to report from inside one, only which of them is running now.	None

Returns:

Type	Description
list[int]	The IDs of the migrations that were applied.

Raises:

Type	Description
DatabaseMigrationError	If a migration fails.

ping async

ping() -> bool

Check whether the database is reachable.

This asks the database a question it can answer without a schema so it says whether the database can be reached rather than whether it has been set up.

Returns:

Type	Description
bool	True if the database answered, False otherwise.

ready async

ready() -> None

Bootstrap an empty database through the migration chain.

Databases that already have tables are left untouched here, assert_schema_current is what guards against stale schemas on those. Bootstrapping only happens once per instance, a cached flag makes every later call zero-I/O. Concurrent first calls may each run initialized() and migrate(), but migrate() serializes on the instance's migration lock so migrations are still only applied once.

Every path that reaches the data has to come through here first, the native store's as much as the query layer's because a database nobody has bootstrapped has no tables and, for a temporary one, no file either.

verify_password async

verify_password(password: str, hash: PasswordHash) -> bool

Check whether a plaintext password matches a stored hash.

If hash is not already a valid password hash it is first hashed with the configured parameters so passing a plaintext value in both arguments verifies the value against itself. The comparison runs on a worker thread.

Parameters:

Name	Type	Description	Default
password	str	Plaintext password to verify.	required
hash	PasswordHash	Stored password hash to compare against.	required

Returns:

Type	Description
bool	True if the password matches the hash, False otherwise.

DatabaseType

Bases: StrEnum

Kind of database backend Ceres is running against.

POSTGRES class-attribute instance-attribute

POSTGRES = 'postgres'

PostgreSQL server, used for shared and production deployments.

SQLITE class-attribute instance-attribute

SQLITE = 'sqlite'

Local SQLite database, the default for single-node deployments.

TURSO class-attribute instance-attribute

TURSO = 'turso'

Local Turso database, a SQLite-compatible backend that allows concurrent writers.

Configuration

Config

Bases: ConfigMeta

Top-level Ceres configuration, including the component tree.

Config is the strict, fully-typed view of a configuration file. Unknown fields are rejected here so users get clear errors for typos. components holds every top-level component configuration, there is no implicit wrapping root component.

access class-attribute instance-attribute

access: ComponentAccessLevel | None = None

Default access level for components with none declared in their ancestor chain.

components class-attribute instance-attribute

components: list[ComponentConfig] = Field(
    default_factory=list
)

Top-level component configurations, the engine's components.

tags class-attribute instance-attribute

tags: list[str] = Field(default_factory=list)

Tags inherited by every component that does not override them.

get_component

get_component(
    address: DynamicAddress,
) -> ComponentConfig | None

Look up a component configuration anywhere in the tree.

get_component_class

get_component_class(
    address: DynamicAddress,
) -> type[Component] | None

Look up the component class declared at address, anywhere in the tree.

get_components

get_components(
    address: AddressSelector | None = None,
) -> dict[Address, ComponentConfig]

Return every component configuration in the tree, optionally filtered.

Parameters:

Name	Type	Description	Default
address	AddressSelector | None	Optional selector restricting which addresses are returned, omit to return every component in the tree.	None

Returns:

Type	Description
dict[Address, ComponentConfig]	Mapping from absolute address to component configuration.

ConfigCheckType

Bases: StrEnum

Identifies optional checks performed during Config.load.

COMPONENTS class-attribute instance-attribute

COMPONENTS = 'components'

Verify the component tree builds without errors.

DATABASE class-attribute instance-attribute

DATABASE = 'database'

Verify the database is reachable and accepts connections.

SERVER class-attribute instance-attribute

SERVER = 'server'

Verify the HTTPS listener's certificate and key load, or for a managed certificate, that its authority can sign one.

all classmethod

all() -> tuple[ConfigCheckType, ...]

Return every defined check, useful as the default for Config.load.

Concurrency

cancel async

cancel(
    *tasks: MaybeRecursiveIterable[Task[Any]],
) -> list[Task[Any]]

Cancel all provided tasks and wait for them to complete.

Exceptions raised during cancellation are suppressed, the goal is to ensure all tasks reach a final state.

Parameters:

Name	Type	Description	Default
tasks	MaybeRecursiveIterable[Task[Any]]	Tasks to cancel, may be passed individually or as nested iterables.	()

Returns:

Type	Description
list[Task[Any]]	A flattened list of all tasks that were cancelled.

concurrently async

concurrently(
    *coroutines: MaybeRecursiveIterable[
        Coroutine[Any, Any, Any] | None
    ]
) -> list[Task[Any]]

Run coroutines concurrently in a task group, waiting for all of them to finish.

Any coroutine that is None is silently ignored, which makes it convenient to conditionally include work without filtering call-site arguments.

Parameters:

Name	Type	Description	Default
coroutines	MaybeRecursiveIterable[Coroutine[Any, Any, Any] | None]	Coroutines to schedule, may be passed individually or as nested iterables.	()

Returns:

Type	Description
list[Task[Any]]	The list of tasks created to run the given coroutines.

sleep async

sleep(delay: float | timedelta | EllipsisType) -> None

Asynchronously sleep for a given delay.

Parameters:

Name	Type	Description	Default
delay	float | timedelta | EllipsisType	A number of seconds, a timedelta representing the duration to sleep for, or ... to sleep forever (or until cancelled).	required

spawn async

spawn(
    function: Callable[P, T], *args: args, **kwargs: kwargs
) -> T

Run a synchronous function in a new thread and await its result.

Parameters:

Name	Type	Description	Default
function	Callable[P, T]	The function to invoke.	required
*args	args	Positional arguments forwarded to function.	()
**kwargs	kwargs	Keyword arguments forwarded to function.	{}

Returns:

Type	Description
T	The value returned by function.

Utilities

utc

utc(value: DateTimeInput | None = None) -> datetime

Get the current time in UTC, or convert a date-time-like value to a UTC datetime.

Parameters:

Name	Type	Description	Default
value	DateTimeInput | None	A datetime, ISO 8601 string, or other accepted date-time input. If None, the current UTC time is returned, honoring any fake-now override set via set_fake_now().	None

Returns:

Type	Description
datetime	A timezone-aware datetime in UTC.

Type Aliases

Each accepts any one of the types listed under it.

DynamicParticleData

Type alias accepting either a raw JSON object or a typed ParticleData instance. Used as the default data annotation on the base Particle class so it can hold either unstructured JSON (as loaded from the database) or a concrete typed payload.

JSONSerializableDict, ParticleData

Entity

Message, Particle, Alert, LogEntry, User, Variable, Setting, Workspace, WorkspaceEdit, Group, GroupMembership, UserPermission, GroupPermission

Item

Union of record types that flow through the system as user-visible items.

Message, Particle, Alert, LogEntry, Variable

MessageData

bytes

Output

FileOutput, StreamingOutput

Record

Union of entities that originate from a component as part of normal data collection. Records are the time-series outputs a component emits during operation, namely raw Message payloads from connections, parsed Particle data, Alert notifications, and LogEntry rows. Workspace, user, and configuration entities are excluded.

Message, Particle[Any], Alert, LogEntry

Schedule

Union of all concrete schedule variants, discriminated by type.

CronSchedule, IntervalSchedule, OrSchedule

StandardEvent

Union of every event type defined in the standard ceres distribution.

StartedEvent, StoppedEvent, EnabledEvent, DisabledEvent, AttachedEvent, WillDetachEvent, DetachedEvent, ConnectionAddedEvent, ConnectionRemovedEvent, ConnectionStartedEvent, ConnectionStoppedEvent, ConnectionExceptionEvent, ConnectedEvent, DisconnectedEvent, DisconnectingEvent, ReceiveTimeoutEvent, DisconnectVerifyStartedEvent, DisconnectVerifiedEvent, DisconnectUnverifiedEvent, DisconnectVerifyEndedEvent, ConnectionLostEvent, ConnectFailedEvent, ReconnectScheduledEvent, BufferOverflowEvent, ServerBindEvent, ServerBindExceptionEvent, ClientConnectedEvent, ClientDisconnectedEvent, ServerProcessingExceptionEvent, MessageSentEvent, MessageReceivedEvent, AlertEvent, LogEvent, RoutineStartedEvent, RoutineStoppedEvent, RoutineCompletedEvent, RoutineExceptionEvent, RoutineRestartedEvent, JobAddedEvent, JobRemovedEvent, JobStartedEvent, JobEndedEvent, JobCompletedEvent, JobCancelledEvent, JobExceptionEvent, JobRetryPendingEvent, JobRetryEvent, PrunerAddedEvent, PrunerRemovedEvent, PruneStartedEvent, PruneEndedEvent, PruneCompletedEvent, PruneCancelledEvent, PruneExceptionEvent, SieveAddedEvent, SieveRemovedEvent, SieveStartedEvent, SieveStoppedEvent, SieveCancelledEvent, SieveExceptionEvent, SieveRetryPendingEvent, SieveRetryEvent, ProcedureCalledEvent, ProcedureCancelledEvent, ProcedureCompletedEvent, ProcedureExceptionEvent, StreamLostEvent, StreamReconnectScheduledEvent, StreamReconnectedEvent, StreamEndedEvent, DatabaseExceptionEvent

ceres.__version__ holds the installed version as a string.