domaindriven 0.1.0

Function-First framework for Domain-Driven Functions — self-describing CapabilityContracts, one definition reachable from every channel, optional durable execution via Temporal.

domaindriven

Function-First framework for Domain-Driven Functions — the DDF pillar of Domain-Driven Enterprise.

A function is the smallest self-describing unit of capability. You declare it once as a CapabilityContract (what it does, its invariants, who may call it, which domain events it emits), and it becomes reachable from every channel — REST, CLI, an LLM tool, a UI, a cron — through one invoke(). The execution substrate is swappable: in-process by default, or durable via Temporal without changing a single caller.

Core has zero required dependencies (stdlib only). YAML and Temporal are optional extras.

Install

pip install domaindriven                 # core
pip install "domaindriven[yaml]"         # + catalog YAML export / round-trip
pip install "domaindriven[temporal]"     # + durable execution via Temporal
pip install "domaindriven[all]"

Hello, function

from domaindriven import FunctionSpec, Param, register, invoke

def _handler(args, ctx):
    return {"greeting": f"Hallo {args['name']}"}

register(FunctionSpec(
    id="demo.greet", description="Grüßt eine Person", category="DEMO",
    returns="greeting", handler=_handler,
    params=[Param(name="name", type="string", required=True)],
    # — optionaler Capability-Contract (DDF/DDI/DDG) —
    bounded_context="Demo", decision="deterministic",
    invariants=("verändert keinen Zustand",), emits=("PersonGreeted",),
))

print(invoke("demo.greet", {"name": "Welt"}))
# {'ok': True, 'greeting': 'Hallo Welt', 'fx': {'id': 'demo.greet', 'worker': 'inline', ...}}

What you get

Modul	Zweck
core	Registry, CapabilityContract (FunctionSpec), register, invoke, Policy (Permission/Params)
worker	austauschbares Ausführungs-Substrat (InlineWorker/ThreadWorker, set_worker)
loader	load_package("meine_app") — Auto-Discovery selbst-registrierender Module
process	ProcessSpec — Prozesse als Komposition von Funktionen (input/action/gate/view)
service	HTTP-Worker (serve([...])) — REST-Kanal für die Registry
catalog	JSON/YAML-Export, Traceability-Matrix, YAML→Contract Round-Trip
schema	JSON-Schema (Editor-Autocomplete) + Scaffold-Template
naming	durchsetzbares Namensschema (Linter)
temporal_adapter	durable execution: TemporalWorker (Function→Activity, Process→Workflow)

Durable execution (optional)

from domaindriven import set_worker
from domaindriven.temporal_adapter import TemporalWorker
from temporalio.client import Client

client = await Client.connect("localhost:7233")
set_worker(TemporalWorker(client, "my-queue"))   # ab jetzt läuft jedes invoke() durable

Function → Activity (Retry/Timeout aus dem Contract), ProcessSpec → Workflow (gate→Signal, view→Query). Der PTDS-Kern passt exakt zu Temporals Determinismus-Grenze: decision="ai_assisted" gehört zwingend in eine Activity.

Status

Teil des dde-Monorepos. Alpha — API kann sich noch ändern.