cloacina-python::constructor
Rust
cloaca.constructor(...) — the Python consumer surface for packaged WASM constructor providers (CLOACI-T-0831), mirroring Rust’s constructor!(...).
Called inside a WorkflowBuilder context (exactly where @cloaca.task
functions are defined), it resolves the named member of a provider suite via
the SAME host loader the Rust surface uses (load_constructor_node:
provider-search-path resolution, name-keyed config binding, name-in-configure
member selection, default-closed capability grants) and registers the resolved
node into the current scoped Runtime under
TaskNamespace(tenant, package, workflow, id) — the workflow’s DAG assembly
then picks it up exactly like a @cloaca.task.
with cloaca.WorkflowBuilder("granted") as _: cloaca.constructor( id="reader", from_="cloacina-provider-fs@0.1.1", constructor="read_file", config={"path": "/data/secret.txt"}, grants={"fs": ["ro:/data"]}, runtime="wasm", # optional trust-tier pin (CLOACI-T-0920) )
@cloaca.task(id="echo", dependencies=["reader"]) def echo(context): ...Execution is language-agnostic — the node runs in the Rust runtime’s WASM
sandbox; Python only authors the wiring. Providers resolve against the same
provider search path as Rust (CLOACINA_PROVIDER_PATH / ./providers /
set_provider_search_path).
Functions
Section titled “Functions”fn constructor
Section titled “fn constructor”Binding
Python API:
cloaca.constructor
fn constructor( py: Python, id: String, from_: String, constructor: String, config: Option<Bound<'_, PyDict>>, grants: Option<Bound<'_, PyDict>>, dependencies: Option<Vec<Py<PyAny>>>, runtime: Option<String>,) -> PyResult<()>Wire a packaged constructor-provider member into the current workflow as a DAG node. See the module docs for the full model; from_ is the provider (Cargo package) name with an optional @version pin, constructor selects the member, config binds by NAME against the member’s declared #[config] schema, and grants is the default-closed capability grant map ({"http": [...], "tcp": [...], "fs": [...], "env": [...]}).
runtime (CLOACI-T-0920) optionally PINS the provider’s trust tier —
"wasm" or "native". If pinned, the load fails unless the resolved
provider’s runtime matches, naming both. Because grants are ENFORCED on wasm
but only ADVISORY on native, passing grants for a provider that resolves
NATIVE is a hard error unless acknowledged with runtime="native".
Source
pub fn constructor( py: Python, id: String, from_: String, constructor: String, config: Option<Bound<'_, PyDict>>, grants: Option<Bound<'_, PyDict>>, dependencies: Option<Vec<Py<PyAny>>>, runtime: Option<String>,) -> PyResult<()> { let context = current_workflow_context()?; let (tenant_id, package_name, workflow_id) = context.as_components(); let namespace = cloacina::TaskNamespace::new(tenant_id, package_name, workflow_id, &id);
// config: dict → (name, value) pairs in WRITTEN order; the loader reorders // them into the member's declaration order and binds by name (kwargs). let mut config_pairs: Vec<(String, serde_json::Value)> = Vec::new(); if let Some(cfg) = &config { for (k, v) in cfg.iter() { let key: String = k.extract().map_err(|_| { PyValueError::new_err("cloaca.constructor: config keys must be strings") })?; let value: serde_json::Value = pythonize::depythonize(&v).map_err(|e| { PyValueError::new_err(format!( "cloaca.constructor: config value for '{key}' is not JSON-convertible: {e}" )) })?; config_pairs.push((key, value)); } }
// grants: {"kind": ["pattern", ...]} → GrantSpec (default-closed when absent). let mut grant_pairs: Vec<(String, Vec<String>)> = Vec::new(); if let Some(g) = &grants { for (k, v) in g.iter() { let kind: String = k.extract().map_err(|_| { PyValueError::new_err( "cloaca.constructor: grant keys must be strings (http/tcp/fs/env)", ) })?; let patterns: Vec<String> = v.extract().map_err(|_| { PyValueError::new_err(format!( "cloaca.constructor: grants['{kind}'] must be a list of strings" )) })?; grant_pairs.push((kind, patterns)); } } let grant_spec = cloacina::registry::loader::grants::GrantSpec::from_pairs(grant_pairs);
// runtime: the optional trust-tier pin. Validated here so a typo is a clear // Python-side error rather than a silent "unpinned" degradation. let runtime_pin = match runtime.as_deref() { None => None, Some("wasm") => Some(cloacina::registry::loader::ProviderRuntime::Wasm), Some("native") => Some(cloacina::registry::loader::ProviderRuntime::Native), Some(other) => { return Err(PyValueError::new_err(format!( "cloaca.constructor: unknown runtime '{other}'; valid values are \ \"wasm\" and \"native\"" ))); } };
// dependencies: strings or function objects (same convention as @task deps). let mut dep_namespaces: Vec<cloacina::TaskNamespace> = Vec::new(); for (i, dep) in dependencies.unwrap_or_default().iter().enumerate() { let dep_id = match dep.extract::<String>(py) { Ok(s) => s, _ => match dep.getattr(py, "__name__") { Ok(name) => name.extract::<String>(py).map_err(|e| { PyValueError::new_err(format!( "cloaca.constructor: dependency {i} __name__ is not a string: {e}" )) })?, _ => { return Err(PyValueError::new_err(format!( "cloaca.constructor: dependency {i} must be a string or a task function" ))); } }, }; dep_namespaces.push(cloacina::TaskNamespace::new( tenant_id, package_name, workflow_id, &dep_id, )); }
// Resolve the member through the host loader (wasmtime compile is blocking // work — release the GIL for it). Fails closed on a missing provider/member, // a bad config kwarg, or a malformed grant. let node = py .detach(|| { cloacina::registry::loader::load_constructor_node_pinned( &id, &from_, &constructor, config_pairs, dep_namespaces, grant_spec, runtime_pin, ) }) .map_err(|e| { PyValueError::new_err(format!( "cloaca.constructor: resolve '{constructor}' from provider '{from_}': {e}" )) })?;
// Register into the current scoped Runtime — the workflow DAG assembly picks // it up exactly like a @cloaca.task registration. let rt = crate::runtime_scope::current_runtime().ok_or_else(|| { PyValueError::new_err( "cloaca.constructor called outside a Runtime scope — install a ScopedRuntime \ before importing Python workflow modules", ) })?; py.detach(|| { rt.register_task(namespace, move || node.clone()); });
Ok(())}