API reference¶

Generated from the source. Everything below is exported from the package root:

from morphly import Config, Delete, Entity, Module, Patch, Put, Step, Store, Workflow, Write, module, view

Business objects¶

The types you subclass to describe your domain.

morphly.Entity ¶

Bases: BaseModel

A shared business object, identified by name within its type lineage.

Subclass it to declare a business object. Identity is the (type, name) pair, and name is frozen: an entity keeps its identity for its whole life, only its other fields change.

Two pydantic settings are enabled: validate_assignment, so every field write is validated — including the ones a Patch applies — and arbitrary_types_allowed, so a field may carry a non-pydantic object such as a dataframe, a matrix or a solver handle.

Subclasses are a first-class case: a module reading list[Employee] also receives every Manager in the store.

Attributes:

Name	Type	Description
`name`	`str`	Stable identifier, unique within the type lineage. Frozen.

Examples:

class Employee(Entity):
    hourly_rate: float
    gross: float = 0.0


class Manager(Employee):
    bonus_target: float

morphly.Config ¶

Bases: BaseModel

A singleton input: module parameters, global settings, context.

A Config has no name because there is only ever one of each type in play. It is resolved by type, either from the Step it is bound to or from the Store.

Examples:

class PayrollPolicy(Config):
    overtime_after: float = 35.0
    overtime_rate: float = 1.25
    social_rate: float = 0.22

Operations¶

What a module returns to declare what it changed.

morphly.Put ¶

Creation marker returned by a module. Annotate as Put[Payslip].

Optional sugar: returning a bare Entity or Config creates or replaces it just the same. Put exists so a signature can spell out the three verbs — Put[Payslip] | Patch[Employee] | Delete[Timesheet] — instead of leaving creation implicit in a bare type sitting in a union.

Like Patch and Delete, it is an intent: nothing is written until the step has produced all of its operations and they have all been validated. The object replaces anything stored under the same (type, name).

Parameters:

Name	Type	Description	Default
`target`	`E`	The entity or config to store.	required

Examples:

@module
def issue(employees: list[Employee]) -> list[Put[Payslip]]:
    return [Put(Payslip(name=e.name, net=e.gross)) for e in employees]

morphly.Patch ¶

Partial update returned by a module. Annotate as Patch[Employee].

A Patch writes only the fields it is given and leaves the rest of the object untouched — the normal output of a module that computes a few attributes on shared objects. Returning a whole Entity, bare or inside a Put, means creation or full replacement instead.

Like Delete, it is an intent: nothing is written until the step has produced all of its operations and they have all been validated.

Parameters:

Name	Type	Description	Default
`target`	`E`	The entity to update, resolved by type lineage and `name`.	required
`**fields`	`Any`	Field names and values to write. At least one is required.	`{}`

Raises:

Type	Description
`ValueError`	If no field is given.

Examples:

@module
def pay(employees: list[Employee]) -> list[Patch[Employee]]:
    return [Patch(e, gross=e.hourly_rate * e.contract_hours) for e in employees]

morphly.Delete ¶

Deletion marker returned by a module. Annotate as Delete[Timesheet].

Returning a Delete does not remove anything by itself: it is an intent, applied by the workflow once the whole step has been validated.

Parameters:

Name	Type	Description	Default
`target`	`E`	The entity to remove. Only its type lineage and `name` are used, so an enriched copy of a stored object is a valid target.	required

Examples:

@module
def archive(timesheets: list[Timesheet]) -> list[Delete[Timesheet]]:
    return [Delete(t) for t in timesheets if t.processed]

morphly.view ¶

view(cls: type[T], source: BaseModel, **extra: Any) -> T

Build an enriched module-local view of a shared object.

Shallow-copies the source fields, adds extra, and validates against cls. Use it when a module needs to carry its own working fields on a shared entity without polluting the shared type with them.

The view can be handed straight back inside a Patch: targets are resolved against the type declared in the return annotation, not the view's own class.

Parameters:

Name	Type	Description	Default
`cls`	`type[T]`	Target model class, usually a subclass of the source class.	required
`source`	`BaseModel`	Object to project.	required
`**extra`	`Any`	Module-specific fields to add.	`{}`

Returns:

Type	Description
`T`	A new `cls` instance holding the source fields plus `extra`.

Raises:

Type	Description
`ValidationError`	If the merged fields do not validate against `cls`.

Examples:

class EmployeeWithHours(Employee):
    worked: float


@module
def pay(employees: list[Employee], sheets: list[Timesheet]) -> list[Patch[Employee]]:
    hours = {s.employee: s.hours for s in sheets}
    rich = [view(EmployeeWithHours, e, worked=hours.get(e.name, 0.0)) for e in employees]
    return [Patch(e, gross=e.hourly_rate * e.worked) for e in rich]

State¶

morphly.Store ¶

The shared state. Buckets by concrete type, read by type (subclasses included).

Entities are keyed by (concrete type, name) and configs by type. Reading is done by type and walks the lineage, so all(Employee) returns the Manager instances too.

The store is a plain Python object: copy.deepcopy(store) is a snapshot and pickle persists it.

Parameters:

Name	Type	Description	Default
`*items`	`Entity \| Config`	Entities and configs to load.	`()`

Examples:

store = Store(
    Employee(name="ada", hourly_rate=50.0, contract_hours=35.0),
    Manager(name="bob", hourly_rate=60.0, contract_hours=35.0, bonus_target=0.1),
    PayrollPolicy(social_rate=0.22),
)
store.all(Employee)  # [Employee(name='ada'), Manager(name='bob')]
store.one(PayrollPolicy).social_rate  # 0.22

put ¶

put(*items: Entity | Config) -> None

Upsert entities (keyed by type and name) and configs (keyed by type).

An entity replaces any object already stored under the same (type, name). For a partial update, use patch instead.

Parameters:

Name	Type	Description	Default
`*items`	`Entity \| Config`	Entities and configs to store.	`()`

Raises:

Type	Description
`TypeError`	If an item is neither an `Entity` nor a `Config`.

all ¶

all(cls: type[E]) -> list[E]

Every instance of cls and of its subclasses.

Parameters:

Name	Type	Description	Default
`cls`	`type[E]`	The entity type to read.	required

Returns:

Type	Description
`list[E]`	The matching entities, in insertion order per bucket. Empty if none.

one ¶

one(cls: type[C]) -> C

The single instance of config cls.

Parameters:

Name	Type	Description	Default
`cls`	`type[C]`	The config type to read.	required

Returns:

Type	Description
`C`	The stored config.

Raises:

Type	Description
`LookupError`	If the store holds zero or several configs of that lineage.

find ¶

find(cls: type[E], name: str) -> E

The object named name: exact type first, then the lineage of cls.

Parameters:

Name	Type	Description	Default
`cls`	`type[E]`	The declared type of the object.	required
`name`	`str`	Its identifier.	required

Returns:

Type	Description
`E`	The stored entity.

Raises:

Type	Description
`KeyError`	If no object matches, or if several sibling types hold that name.

drop ¶

drop(
    target: Entity | type[Entity], name: str | None = None
) -> None

Remove the stored object matching target.

Parameters:

Name	Type	Description	Default
`target`	`Entity \| type[Entity]`	An entity with the type lineage and `name` to remove — it does not have to be the stored instance itself — or the entity type, paired with `name`.	required
`name`	`str \| None`	The object's name, when `target` is a type.	`None`

Raises:

Type	Description
`TypeError`	If `target` is a type and `name` is not given.
`KeyError`	If no object matches, or if the name is ambiguous.

patch ¶

patch(
    target: Entity | type[Entity],
    name_or_fields: str | dict[str, Any],
    fields: dict[str, Any] | None = None,
) -> None

Write fields onto the stored object matching target.

Parameters:

Name	Type	Description	Default
`target`	`Entity \| type[Entity]`	An entity with the type lineage and `name` to update, or the entity type, paired with a name and fields.	required
`name_or_fields`	`str \| dict[str, Any]`	Field names and values to write, when `target` is an entity, else the object's name.	required
`fields`	`dict[str, Any] \| None`	Field names and values to write, when `target` is a type.	`None`

Raises:

Type	Description
`TypeError`	If `target` is a type and `fields` is not given.
`KeyError`	If no object matches, or if the name is ambiguous.
`ValueError`	If a name is not a field of the stored object.
`ValidationError`	If a value does not validate.

history ¶

history(
    target: Entity | type[Entity], name: str | None = None
) -> list[Write]

Every write a workflow made to that object, oldest first.

The runtime counterpart of Workflow.to_mermaid: the graph says which steps may write a type, the history says which ones did. Recorded by Workflow.run, so a write made by calling put/patch/drop directly leaves no entry, and configs are not tracked — they have no name to key them by.

The log outlives the object: a deleted entity keeps its history, ending with its delete. Reads follow the lineage like all rather than find, so an ambiguous name returns the writes of every matching sibling type instead of raising.

Parameters:

Name	Type	Description	Default
`target`	`Entity \| type[Entity]`	An entity with the type lineage and `name` to look up, or the entity type, paired with `name`.	required
`name`	`str \| None`	The object's name, when `target` is a type.	`None`

Returns:

Type	Description
`list[Write]`	The recorded writes. Empty if the object was never written by a workflow.

Examples:

store.history(payslip)
# [Write(step='issue_payslip', action='put', fields=()),
#  Write(step='withhold_tax', action='patch', fields=('net',))]

types ¶

types() -> set[type]

Types currently present, entities and configs.

Returns:

Type	Description
`set[type]`	The concrete types held by the store. This is the starting point of
`set[type]`	`Workflow.check`.

morphly.Write ¶

Bases: NamedTuple

One recorded write, as returned by history.

Attributes:

Name	Type	Description
`step`	`str`	Name of the step that wrote.
`action`	`str`	`"put"`, `"patch"` or `"delete"` — the operation the module returned, not its effect, so a `put` that replaced an existing object still reads `"put"`.
`fields`	`tuple[str, ...]`	The fields a `patch` wrote. Empty for `put` and `delete`.

Orchestration¶

morphly.module ¶

Modules and steps: the function signature is the contract.

Module ¶

A function whose annotations declare what it reads, creates and touches.

Built by the @module decorator; you rarely instantiate it yourself. The annotations are parsed once, at import time, and an unsupported or missing one fails there rather than mid-run.

Annotation	Injected
`list[X]`, `X: Entity`	`store.all(X)`, subclasses included
`X`, `X: Config`	the step's config, else `store.one(X)`

The return annotation forms the output contract: Entity and Config types, bare or inside Put[...], are produced, types inside Patch[...] and Delete[...] are touched, and None means the module writes nothing.

Parameters:

Name	Type	Description	Default
`fn`	`Callable[..., Any]`	The annotated function to wrap.	required

Attributes:

Name	Type	Description
`fn`		The wrapped function.
`name`		The function's name, used as the default step name.
`reads`	`list[tuple[str, bool, type]]`	One `(parameter, is_collection, type)` triple per parameter.
`produces`	`list[tuple[str, bool, type]]`	Entity and config types the module creates or replaces.
`touches`	`list[tuple[str, bool, type]]`	Entity types the module patches or deletes.

Raises:

Type	Description
`TypeError`	If a parameter is unannotated, if an annotation is not `list[Entity subclass]` or a `Config` subclass, if the return is unannotated, or if `Put`/`Patch`/`Delete` is used without a type parameter.

call ¶

__call__(
    store: Store,
    configs: tuple[Config, ...] = (),
    copy_inputs: bool = True,
) -> list[_Op]

Run the function, returning its operations without applying them.

Useful on its own to unit-test a module: build a Store, call the module, and assert on the operations it returns — nothing is written.

Parameters:

Name	Type	Description	Default
`store`	`Store`	The state to read from.	required
`configs`	`tuple[Config, ...]`	Configs bound to this step, tried before `store.one`.	`()`
`copy_inputs`	`bool`	Deep-copy every injected value, so the module cannot corrupt the state read by another one. Turn off only when volume demands it.	`True`

Returns:

Type	Description
`list[_Op]`	The operations the module returned, validated against its contract but not
`list[_Op]`	yet applied. Empty if the module returned `None`.

Raises:

Type	Description
`TypeError`	If the module returned an operation its signature does not declare.
`LookupError`	If a required config is neither bound to the step nor in the store.

Step ¶

A module plus the configs bound to that occurrence of it.

Configs live on the step, not on the module, so the same module can appear twice in a workflow with different parameters. A config bound here wins over the one in the Store.

Parameters:

Name	Type	Description	Default
`module_`	`Module \| Callable[..., Any]`	A `Module`, or a plain function wrapped on the fly.	required
`*configs`	`Config`	Configs bound to this occurrence.	`()`
`name`	`str \| None`	Step name for `explain` and `on_step`. Defaults to the function's name, suffixed on collision within a workflow.	`None`

Attributes:

Name	Type	Description
`module`		The wrapped module.
`configs`		The configs bound to this step.
`name`		The step name.

Examples:

Workflow(
    Step(withhold, PayrollPolicy(social_rate=0.22), name="withhold_fr"),
    Step(withhold, PayrollPolicy(social_rate=0.13), name="withhold_uk"),
)

module ¶

module(fn: Callable[..., Any]) -> Module

Turn an annotated function into a module.

There is no base class to inherit from and nothing to register: the decorator reads the annotations and that is the whole contract. Errors in the contract are raised at import time.

Parameters:

Name	Type	Description	Default
`fn`	`Callable[..., Any]`	A function whose parameters are annotated `list[Entity subclass]` or `Config subclass`, and whose return type is annotated.	required

Returns:

Type	Description
`Module`	A `Module`, ready to be dropped into a
`Module`	`Workflow`.

Raises:

Type	Description
`TypeError`	If the signature does not form a valid contract.

Examples:

@module
def compute_gross(
    employees: list[Employee],
    sheets: list[Timesheet],
    policy: PayrollPolicy,
) -> list[Patch[Employee]]:
    hours = {s.employee: s.hours for s in sheets}
    return [Patch(e, gross=e.hourly_rate * hours.get(e.name, 0.0)) for e in employees]

morphly.Module ¶

A function whose annotations declare what it reads, creates and touches.

Built by the @module decorator; you rarely instantiate it yourself. The annotations are parsed once, at import time, and an unsupported or missing one fails there rather than mid-run.

Annotation	Injected
`list[X]`, `X: Entity`	`store.all(X)`, subclasses included
`X`, `X: Config`	the step's config, else `store.one(X)`

The return annotation forms the output contract: Entity and Config types, bare or inside Put[...], are produced, types inside Patch[...] and Delete[...] are touched, and None means the module writes nothing.

Parameters:

Name	Type	Description	Default
`fn`	`Callable[..., Any]`	The annotated function to wrap.	required

Attributes:

Name	Type	Description
`fn`		The wrapped function.
`name`		The function's name, used as the default step name.
`reads`	`list[tuple[str, bool, type]]`	One `(parameter, is_collection, type)` triple per parameter.
`produces`	`list[tuple[str, bool, type]]`	Entity and config types the module creates or replaces.
`touches`	`list[tuple[str, bool, type]]`	Entity types the module patches or deletes.

Raises:

Type	Description
`TypeError`	If a parameter is unannotated, if an annotation is not `list[Entity subclass]` or a `Config` subclass, if the return is unannotated, or if `Put`/`Patch`/`Delete` is used without a type parameter.

call ¶

__call__(
    store: Store,
    configs: tuple[Config, ...] = (),
    copy_inputs: bool = True,
) -> list[_Op]

Run the function, returning its operations without applying them.

Useful on its own to unit-test a module: build a Store, call the module, and assert on the operations it returns — nothing is written.

Parameters:

Name	Type	Description	Default
`store`	`Store`	The state to read from.	required
`configs`	`tuple[Config, ...]`	Configs bound to this step, tried before `store.one`.	`()`
`copy_inputs`	`bool`	Deep-copy every injected value, so the module cannot corrupt the state read by another one. Turn off only when volume demands it.	`True`

Returns:

Type	Description
`list[_Op]`	The operations the module returned, validated against its contract but not
`list[_Op]`	yet applied. Empty if the module returned `None`.

Raises:

Type	Description
`TypeError`	If the module returned an operation its signature does not declare.
`LookupError`	If a required config is neither bound to the step nor in the store.

morphly.Step ¶

A module plus the configs bound to that occurrence of it.

Configs live on the step, not on the module, so the same module can appear twice in a workflow with different parameters. A config bound here wins over the one in the Store.

Parameters:

Name	Type	Description	Default
`module_`	`Module \| Callable[..., Any]`	A `Module`, or a plain function wrapped on the fly.	required
`*configs`	`Config`	Configs bound to this occurrence.	`()`
`name`	`str \| None`	Step name for `explain` and `on_step`. Defaults to the function's name, suffixed on collision within a workflow.	`None`

Attributes:

Name	Type	Description
`module`		The wrapped module.
`configs`		The configs bound to this step.
`name`		The step name.

Examples:

Workflow(
    Step(withhold, PayrollPolicy(social_rate=0.22), name="withhold_fr"),
    Step(withhold, PayrollPolicy(social_rate=0.13), name="withhold_uk"),
)

morphly.Workflow ¶

An ordered list of steps, checked before it runs.

The order is yours: a workflow is a list, not a scheduler. What it does guarantee is that an inconsistent order — a step reading a type nobody upstream provides — fails in a millisecond instead of three hours in.

Parameters:

Name	Type	Description	Default
`*steps`	`Step \| Module \| Callable[..., Any]`	`Step` instances, modules, or plain functions. A bare module is equivalent to `Step(module)`.	`()`

Attributes:

Name	Type	Description
`steps`		The steps, in order, with duplicate names suffixed (`pay`, `pay_2`).
`last_run`	`_RunCache \| None`	Cache of the most recent `run(record=True)`, in memory. Pass it as `reuse` to a later `run` to skip steps whose inputs haven't changed. `None` until a run records one.

Examples:

workflow = Workflow(
    Step(compute_gross, PayrollPolicy(overtime_after=35.0)),
    withhold,
    archive_timesheets,
)
workflow.run(store)

check ¶

check(store: Store) -> None

Validate the chaining on types only, without running anything.

Replays the workflow on types: it starts from store.types() and adds, after each step, the types that step produces. A step reading or touching a type that is neither in the initial store nor produced upstream is an error.

Called automatically by run; call it yourself to fail at startup, before loading any data.

A type that is produced, but by a step further down the list, is reported as an ordering problem, with the position to move: the workflow is complete, only its order is wrong, and that is the far more common mistake.

Parameters:

Name	Type	Description	Default
`store`	`Store`	The state the workflow would run on. Only its types are read.	required

Raises:

Type	Description
`LookupError`	If a step reads or touches a type nobody provides.

run ¶

run(
    store: Store,
    *,
    copy_inputs: bool = True,
    on_step: Callable[[Step, list[_Op], Store], None]
    | None = None,
    reuse: _RunCache | None = None,
    record: bool = False,
    atomic: bool = False,
) -> Store

Check, then run every step in order and apply its output.

A step's operations are collected and validated before any of them is written, so a step never applies halfway. The workflow as a whole is transactional only with atomic=True.

An exception raised by a module, or while applying its output, is re-raised unchanged with a note naming the step, its position and the state of the store — the context you want when step 12 of 40 fails three hours into a batch.

Parameters:

Name	Type	Description	Default
`store`	`Store`	The state to run on. Mutated in place.	required
`copy_inputs`	`bool`	Deep-copy the values injected into each module. Turning this off drops the isolation guarantee — a module mutating an input then affects the shared state.	`True`
`on_step`	`Callable[[Step, list[_Op], Store], None] \| None`	Called as `on_step(step, ops, store)` after each step is applied, with the operations it just wrote. The hook for logs, metrics, provenance and writing intermediate outputs.	`None`
`reuse`	`_RunCache \| None`	A previous [`last_run`][morphly.Workflow.last_run] to replay from. A step whose reads are byte-for-byte identical to that run is skipped — its recorded outcome is restored from an in-memory snapshot instead of calling the module again. The first step whose reads differ, and every step after it, runs for real. Handy in a notebook: touch step 12, rerun, the first 11 are skipped.	`None`
`record`	`bool`	Populate [`last_run`][morphly.Workflow.last_run], so a later run can `reuse` it. Off by default because it is not free: it snapshots the store after every step and serialises everything every step reads, so a recorded run costs one deep copy of the store per step in time and in memory. Turn it on where that pays for itself — a notebook, an interactive session — not in a batch that will never be replayed.	`False`
`atomic`	`bool`	Restore the store to its exact initial state if any step raises, making the whole run all-or-nothing instead of just each step. Costs one deep copy of the store, taken once before the first step.	`False`

Returns:

Type	Description
`Store`	The same `store`, mutated.

Raises:

Type	Description
`LookupError`	If `check` fails, or a config is missing.
`TypeError`	If a module returns an operation it did not declare.
`KeyError`	If a `Patch` or `Delete` target is missing or ambiguous.
`ValueError`	If a `Patch` names a field the target does not have, or if a step returns an operation on a target it already deleted.
`ValidationError`	If a written value does not validate.

explain ¶

explain() -> str

One line per step: reads(), then produced and ~touched types.

Returns:

Type	Description
`str`	A rendering of the workflow's dataflow, for logs and reviews. Touched types
`str`	are prefixed with `~`.

Examples:

1. compute_gross: compute_gross(Employee[], Timesheet[], PayrollPolicy) -> ~Employee
2. withhold: withhold(Employee[], PayrollPolicy) -> Payslip
3. archive: archive(Timesheet[]) -> ~Timesheet

to_mermaid ¶

to_mermaid() -> str

Render the workflow's dataflow as a Mermaid flowchart.

Same information as explain, as a graph instead of a line per step: every edge is a reads/produces/touches relationship already present in the modules' signatures, so the graph cannot go stale.

Returns:

Type	Description
`str`	A `flowchart LR` block: `Type --> step` for entity reads, `step --> Type`
`str`	for produced types, `step -.-> Type` for touched types.

Examples:

print(workflow.to_mermaid())

flowchart LR
    Employee --> compute_gross
    Timesheet --> compute_gross
    compute_gross -.-> Employee
    Employee --> withhold
    withhold --> Payslip

API reference¶

Business objects¶

morphly.Entity ¶

morphly.Config ¶

Operations¶

morphly.Put ¶

morphly.Patch ¶

morphly.Delete ¶

morphly.view ¶

State¶

morphly.Store ¶

put ¶

all ¶

one ¶

find ¶

drop ¶

patch ¶

history ¶

types ¶

morphly.Write ¶

Orchestration¶

morphly.module ¶

Module ¶

__call__ ¶

Step ¶

module ¶

morphly.Module ¶

__call__ ¶

morphly.Step ¶

morphly.Workflow ¶

check ¶

run ¶

explain ¶

to_mermaid ¶

call ¶

call ¶