cloacina-python::context

Rust

Structs

class Context

Binding

Python API: cloaca.Context

PyContext - Python wrapper for Rust Context<serde_json::Value>

Fields

Name	Type	Description
inner	cloacina::Context<serde_json::Value>

Methods

new

fn new(data: Option<&Bound<'_, PyDict>>) -> PyResult<Self>

Python API: cloaca.Context.new

Creates a new empty context

Source

    pub fn new(data: Option<&Bound<'_, PyDict>>) -> PyResult<Self> {
        let mut context = cloacina::Context::new();

        if let Some(dict) = data {
            for (key, value) in dict.iter() {
                let key_str: String = key.extract()?;
                let json_value: serde_json::Value = depythonize(&value)?;
                context.insert(key_str, json_value).map_err(|e| {
                    PyErr::new::<pyo3::exceptions::PyValueError, _>(format!(
                        "Failed to insert key: {}",
                        e
                    ))
                })?;
            }
        }

        Ok(PyContext { inner: context })
    }

get

fn get(
    &self,
    key: &str,
    default: Option<&Bound<'_, PyAny>>,
) -> PyResult<Py<PyAny>>

Python API: cloaca.Context.get

Gets a value from the context

Source

    pub fn get(&self, key: &str, default: Option<&Bound<'_, PyAny>>) -> PyResult<Py<PyAny>> {
        match self.inner.get(key) {
            Some(value) => Python::attach(|py| Ok(pythonize(py, value)?.into())),
            None => match default {
                Some(default_value) => Ok(default_value.clone().into()),
                None => Python::attach(|py| Ok(py.None())),
            },
        }
    }

set

fn set(&mut self, key: &str, value: &Bound<'_, PyAny>) -> PyResult<()>

Python API: cloaca.Context.set

Sets a value in the context (insert or update)

Source

    pub fn set(&mut self, key: &str, value: &Bound<'_, PyAny>) -> PyResult<()> {
        let json_value: serde_json::Value = depythonize(value)?;

        if self.inner.get(key).is_some() {
            self.inner.update(key, json_value)
        } else {
            self.inner.insert(key, json_value)
        }
        .map_err(|e| {
            PyErr::new::<pyo3::exceptions::PyValueError, _>(format!(
                "Failed to set key '{}': {}",
                key, e
            ))
        })
    }

update

fn update(
    &mut self) -> PyResult<(,
    key: &str) -> PyResult<(,
    value: &Bound<'_, PyAny>) -> PyResult<(,
) -> PyResult<()>

Python API: cloaca.Context.update

Updates an existing value in the context

Source

    pub fn update(&mut self, key: &str, value: &Bound<'_, PyAny>) -> PyResult<()> {
        let json_value: serde_json::Value = depythonize(value)?;
        self.inner.update(key, json_value).map_err(|e| {
            PyErr::new::<pyo3::exceptions::PyKeyError, _>(format!("Key not found: {}", e))
        })
    }

insert

fn insert(
    &mut self) -> PyResult<(,
    key: &str) -> PyResult<(,
    value: &Bound<'_, PyAny>) -> PyResult<(,
) -> PyResult<()>

Python API: cloaca.Context.insert

Inserts a new value into the context

Source

    pub fn insert(&mut self, key: &str, value: &Bound<'_, PyAny>) -> PyResult<()> {
        let json_value: serde_json::Value = depythonize(value)?;
        self.inner.insert(key, json_value).map_err(|e| {
            PyErr::new::<pyo3::exceptions::PyValueError, _>(format!("Key already exists: {}", e))
        })
    }

remove

fn remove(&mut self, key: &str) -> PyResult<Option<Py<PyAny>>>

Python API: cloaca.Context.remove

Removes and returns a value from the context

Source

    pub fn remove(&mut self, key: &str) -> PyResult<Option<Py<PyAny>>> {
        match self.inner.remove(key) {
            Some(value) => Python::attach(|py| Ok(Some(pythonize(py, &value)?.into()))),
            None => Ok(None),
        }
    }

to_​dict

fn to_dict(&self, py: Python<'_>) -> PyResult<Py<PyAny>>

Python API: cloaca.Context.to_dict

Returns the context as a Python dictionary

Source

    pub fn to_dict(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
        Ok(pythonize(py, self.inner.data())?.into())
    }

update_​from_​dict

fn update_from_dict(&mut self, data: &Bound<'_, PyDict>) -> PyResult<()>

Python API: cloaca.Context.update_from_dict

Updates the context with values from a Python dictionary

Source

    pub fn update_from_dict(&mut self, data: &Bound<'_, PyDict>) -> PyResult<()> {
        for (key, value) in data.iter() {
            let key_str: String = key.extract()?;
            let json_value: serde_json::Value = depythonize(&value)?;

            if self.inner.get(&key_str).is_some() {
                self.inner.update(key_str, json_value)
            } else {
                self.inner.insert(key_str, json_value)
            }
            .map_err(|e| {
                PyErr::new::<pyo3::exceptions::PyValueError, _>(format!(
                    "Failed to update from dict: {}",
                    e
                ))
            })?;
        }
        Ok(())
    }

to_​json

fn to_json(&self) -> PyResult<String>

Python API: cloaca.Context.to_json

Serializes the context to a JSON string

Source

    pub fn to_json(&self) -> PyResult<String> {
        self.inner.to_json().map_err(|e| {
            PyErr::new::<pyo3::exceptions::PyValueError, _>(format!(
                "Failed to serialize to JSON: {}",
                e
            ))
        })
    }

from_​json

fn from_json(json_str: &str) -> PyResult<Self>

Python API: cloaca.Context.from_json

Creates a context from a JSON string

Source

    pub fn from_json(json_str: &str) -> PyResult<Self> {
        let context = cloacina::Context::from_json(json_str.to_string()).map_err(|e| {
            PyErr::new::<pyo3::exceptions::PyValueError, _>(format!(
                "Failed to deserialize from JSON: {}",
                e
            ))
        })?;
        Ok(PyContext { inner: context })
    }

_​_​len_​_​

fn __len__(&self) -> usize

Python API: cloaca.Context.__len__

Returns the number of key-value pairs in the context

Source

    pub fn __len__(&self) -> usize {
        self.inner.data().len()
    }

_​_​contains_​_​

fn __contains__(&self, key: &str) -> bool

Python API: cloaca.Context.__contains__

Checks if a key exists in the context

Source

    pub fn __contains__(&self, key: &str) -> bool {
        self.inner.get(key).is_some()
    }

_​_​repr_​_​

fn __repr__(&self) -> String

Python API: cloaca.Context.__repr__

String representation of the context

Source

    pub fn __repr__(&self) -> String {
        match self.inner.to_json() {
            Ok(json) => format!("Context({})", json),
            Err(_) => "Context(<serialization error>)".to_string(),
        }
    }

_​_​getitem_​_​

fn __getitem__(&self, key: &str) -> PyResult<Py<PyAny>>

Python API: cloaca.Context.__getitem__

Dictionary-style item access

Source

    pub fn __getitem__(&self, key: &str) -> PyResult<Py<PyAny>> {
        let result = self.get(key, None)?;
        Python::attach(|py| {
            if result.is_none(py) {
                Err(PyErr::new::<pyo3::exceptions::PyKeyError, _>(format!(
                    "Key not found: '{}'",
                    key
                )))
            } else {
                Ok(result)
            }
        })
    }

_​_​setitem_​_​

fn __setitem__(
    &mut self) -> PyResult<(,
    key: &str) -> PyResult<(,
    value: &Bound<'_, PyAny>) -> PyResult<(,
) -> PyResult<()>

Python API: cloaca.Context.__setitem__

Dictionary-style item assignment

Source

    pub fn __setitem__(&mut self, key: &str, value: &Bound<'_, PyAny>) -> PyResult<()> {
        self.set(key, value)
    }

_​_​delitem_​_​

fn __delitem__(&mut self, key: &str) -> PyResult<()>

Python API: cloaca.Context.__delitem__

Dictionary-style item deletion

Source

    pub fn __delitem__(&mut self, key: &str) -> PyResult<()> {
        match self.remove(key)? {
            Some(_) => Ok(()),
            None => Err(PyErr::new::<pyo3::exceptions::PyKeyError, _>(format!(
                "Key not found: '{}'",
                key
            ))),
        }
    }

secret

fn secret(&self, py: Python<'_>, name: &str) -> PyResult<Py<PyAny>>

Python API: cloaca.Context.secret

Read a resolved secret by name, returning a dict of its {field: value} pairs (CLOACI-I-0133 / T-0864 — the Python read accessor).

Calls through to the SAME Rust cloacina::Context::secret accessor and the SAME SecretResolver the runtime attached to this context at execution time, so it is alias-aware through the instance’s {"$secret"} map (T-0859): name may be either the declared local binding name or the concrete secret name.

No-leak: the resolved value is returned to the caller ONLY. It is never written back into the context’s serialized data, so it cannot reach schedules.params, the fires log, or execution history.

Raises RuntimeError when no resolver is configured on this execution scope (CLOACINA_SECRET_KEK unset) or the backend fails, KeyError when the secret is not found, PermissionError when the name is not granted.

Source

    pub fn secret(&self, py: Python<'_>, name: &str) -> PyResult<Py<PyAny>> {
        let fields = crate::gil::py_block_on(py, self.inner.secret(name))
            .map_err(secret_access_error_to_pyerr)?;
        let dict = PyDict::new(py);
        for (field, value) in fields {
            dict.set_item(field, value)?;
        }
        Ok(dict.into())
    }

secret_​field

fn secret_field(
    &self,
    py: Python<'_>,
    name: &str,
    field: &str,
) -> PyResult<String>

Python API: cloaca.Context.secret_field

Read a single field of a named secret (convenience over Self::secret).

Same resolver, alias-awareness, no-leak, and error mapping as Self::secret; additionally raises KeyError when the secret exists but has no field of that name.

Source

    pub fn secret_field(&self, py: Python<'_>, name: &str, field: &str) -> PyResult<String> {
        crate::gil::py_block_on(py, self.inner.secret_field(name, field))
            .map_err(secret_access_error_to_pyerr)
    }

from_​rust_​context

pub

fn from_rust_context(
    context: cloacina::Context<serde_json::Value>,
) -> Self

Create a PyContext from a Rust Context (for internal use)

Source

    pub fn from_rust_context(context: cloacina::Context<serde_json::Value>) -> Self {
        PyContext { inner: context }
    }

into_​inner

pub

fn into_inner(self) -> cloacina::Context<serde_json::Value>

Extract the inner Rust Context (for internal use)

Source

    pub fn into_inner(self) -> cloacina::Context<serde_json::Value> {
        self.inner
    }

clone_​inner

pub

fn clone_inner(&self) -> cloacina::Context<serde_json::Value>

Clone the inner Rust Context (for internal use)

Source

    pub fn clone_inner(&self) -> cloacina::Context<serde_json::Value> {
        self.inner.clone_data()
    }

get_​data_​clone

pub

fn get_data_clone(
    &self,
) -> std::collections::HashMap<String, serde_json::Value>

Get a clone of the context data as a HashMap (for internal use)

Source

    pub fn get_data_clone(&self) -> std::collections::HashMap<String, serde_json::Value> {
        self.inner.data().clone()
    }