UniteLabs
package

unitelabs.sdk

unitelabs/sdk/__init__.py

Packages

Attributes

  • __version__
    = version('unitelabs_sdk')
  • __all__
    = [ "unitelabs.sdk.client.client.AsyncApiClient", "unitelabs.sdk.automate.audit.Audit", "unitelabs.sdk.client.client.Client", "unitelabs.sdk.automate.context.Context", "unitelabs.sdk.automate.context.ExecutionContext", "unitelabs.sdk.automate.audit.Operation", "unitelabs.sdk.automate.hitl.OperatorConfirmation", "unitelabs.sdk.automate.context.RuntimeContext", "unitelabs.sdk.client.client.SyncApiClient", "unitelabs.sdk.automate.context.get_context", "unitelabs.sdk.automate.logging.get_logger", "unitelabs.sdk.automate.hitl.operator_confirm", "unitelabs.sdk.automate.decorators.phase", "unitelabs.sdk.automate.decorators.step", "unitelabs.sdk.automate.decorators.workflow" ]

Classes

  • Audit

    class

    Track records of all entities and the operations performed on them.

    Methods

    • @classmethod
      context(cls, context : dict) -> None

      Set the overall context to the given object.

      context
      dict
    • @classmethod
      emit(
        cls, 
        actor : str, 
        inputs : dict, 
        outputs : dict, 
        version : str, 
        event_type : typing.Literal['lineage', 'audit', 'identification', 'measurement'] | str, 
        extras : dict | None
      ) -> None

      Register a new operation. Emits a no-op when called outside an active context so lineage failures never break a running workflow. When `prefect.runtime` is available, the active task and flow run identifiers are attached to the event.

      actor
      str
      The device or service performing the operation.
      operation
      Operation
      The operation being performed.
      inputs
      dict
      Entities consumed by the operation.
      outputs
      dict
      Entities produced by the operation.
      version
      str = '1.0.0'
      Schema version for the event payload.
      event_type
      typing.Literal['lineage', 'audit', 'identification', 'measurement'] | str = 'lineage'
      Stream discriminator. Standard values are `"lineage"`, `"audit"`, `"identification"`, `"measurement"`. Workflow authors may extend with custom values; downstream consumers branch on this field to route events. Defaults to `"lineage"`.
      extras
      dict | None = None
      Optional per-event metadata (e.g., volume, channel, run mode).

  • Context

    class

    Manages the active ExecutionContext across workflow boundaries.

    Methods

    • @classmethod
      @contextlib.contextmanager
      provide(cls, context : RuntimeContext) -> collections.abc.Generator[None, None, None]

      Activate context for the duration of the block.

      context
      RuntimeContext
      The `ExecutionContext` to make active.
      collections.abc.Generator[None, None, None]
    • @classmethod
      @contextlib.asynccontextmanager
      get_or_load(cls, path : pathlib.Path) -> collections.abc.AsyncGenerator[RuntimeContext, None]

      Yield the active context, loading from disk if none is set. Saves context state on exit for Prefect flow-boundary persistence.

      path
      pathlib.Path = _DEFAULT_PATH
      Filesystem path for context persistence.
      collections.abc.AsyncGenerator[RuntimeContext, None]
    • @classmethod
      load(cls, path : pathlib.Path) -> RuntimeContext

      Load the context from the given location.

      path
      pathlib.Path = _DEFAULT_PATH
      RuntimeContext
    • @classmethod
      save(cls, path : pathlib.Path) -> None

      Save the context to the given location.

      path
      pathlib.Path = _DEFAULT_PATH

  • ExecutionContext

    class

    Holds serializable workflow execution state.

    Bases
    typing.Generic[T]

    Methods

    • __init__(self, initial_state : T | None) -> None

      initial_state
      T | None = None
    • @classmethod
      deserialize(cls, data : bytes) -> typing.Self

      Reconstruct context from serialized bytes.

      data
      bytes
      JSON-encoded bytes from serialize().
      typing.Self
      A new ExecutionContext with the deserialized state.
    • serialize(self) -> bytes

      Serialize context state to a versioned JSON envelope. State keys are written at the top level alongside `schema_version`. Consumers read events directly via `payload["events"]`, metadata via `payload["run_id"]`, etc. — no nested unwrapping needed.

      bytes
      UTF-8 encoded JSON bytes of the versioned envelope.

    Attributes

    • state
      = initial_state or {}

  • Operation

    class

    An operation performed on an entity.

    Bases
    enum.Enum

    Attributes

    • ASPIRATE
      = 'aspirate'
    • DISPENSE
      = 'dispense'
    • PICK_UP_TIP
      = 'pick_up_tip'
    • PUT_DOWN_TIP
      = 'put_down_tip'
    • DISCARD_TIP
      = 'discard_tip'
    • MOVE_LABWARE
      = 'move_labware'
    • LOAD_LABWARE
      = 'load_labware'
    • UNLOAD_LABWARE
      = 'unload_labware'

  • OperatorConfirmation

    class

    Schema sent to the UI so the Input tab renders a resume button.

    Bases
    prefect.input.run_input.RunInput

    Attributes

    • confirmed
      bool = True

  • RuntimeContext

    class

    Execution context available within a running workflow.

    Methods

    • __init__(self, initial_state : T | None) -> None

      initial_state
      T | None = None

    Attributes

    • _client
      AsyncApiClient | None = None
    • devices
      dict = {}
    • client
      AsyncApiClient = None
      A client instance for direct communication with the UniteLabs API.

  • AsyncApiClient

    class

    Asynchronous API client for UniteLabs services.

    MRO

  • Client

    class

    Deprecated: Use AsyncApiClient instead.

    MRO
    • └────────── Client

    Methods

    • __init__(self, *args, **kwargs) -> None

      *args
      = ()
      **kwargs
      = {}

  • SyncApiClient

    class

    Synchronous API client for UniteLabs services.

    MRO