Memory¶

class laminar.configurations.datastores.Memory(root: str = 'memory:///', cache: ~typing.Dict[str, ~typing.Any] = <factory>, protocols: ~typing.Dict[str, ~laminar.configurations.serde.Protocol] = <factory>)¶

Bases: DataStore

Store the laminar workspace in memory.

Usage:

Flow(datastore=Memory())

Methods

`exists`	Check if a file exists in the datastore.
`list_artifacts`	List all artifacts in a layer execution.
`list_executions`	List all executions.
`list_layers`	List all layers in an execution.
`protocol`	Register a custom serde protocol for a type.
`read`	Read from the laminar datastore.
`read_archive`	Read an archive from the laminar datastore.
`read_artifact`	Read an artifact form the laminar datastore.
`read_record`	Read a layer record from the laminar datastore.
`uri`	Given a path, generate a URI in the datastore.
`write`	Write to the laminar datastore.
`write_archive`	Write an archive to the laminar datastore.
`write_artifact`	Write an arifact to the laminar datastore.
`write_record`	Write a layer record to the laminar datastore.

Attributes

`root`	URI root of the datastore

__init__(root: str = 'memory:///', cache: ~typing.Dict[str, ~typing.Any] = <factory>, protocols: ~typing.Dict[str, ~laminar.configurations.serde.Protocol] = <factory>)¶

cache : dict[str, Any]¶: Internal datastore cache

exists(*, path: str) → bool¶

Check if a file exists in the datastore.

Parameters¶

path: str¶: Path from the datastore root to the file.

Returns¶

True if the file exists else False

list_artifacts(*, layer: Layer) → list[str]¶

List all artifacts in a layer execution.

Parameters¶

layer: Layer¶: Layer to list artifacts for.

Returns¶

All artifacts.

list_executions(*, flow: Flow) → list[Execution]¶

List all executions.

Parameters¶

flow: Flow¶: Flow to list executions for.

Returns¶

All executions.

list_layers(*, execution: Execution) → list[Layer]¶

List all layers in an execution.

Parameters¶

execution: Execution¶: Execution to list layers for.

Returns¶

All layers.

protocol(*dtypes: type) → Callable[[type[ProtocolType]], type[ProtocolType]]¶

Register a custom serde protocol for a type.

Usage:

@datastore.protocol(pd.DataFrame)
def DataFrameProtocol(serde.Protocol):
    ...

protocols : dict[str, Protocol]¶: Custom serde protocols for reading/writing artifacts

read(*, layer: Layer, index: int, name: str) → Any¶

Read from the laminar datastore.

Parameters¶

layer: Layer¶: Layer being read from.
index: int¶: Layer index being read from.
name: str¶: Name of the artifact being read.

Returns¶

Value of the artifact.

Return type¶

Any

read_archive(*, layer: Layer, index: int, name: str, cache: bool = False) → Archive¶

Read an archive from the laminar datastore.

Parameters¶

layer: Layer¶: Layer being read from.
index: int¶: Layer index being read from.
name: str¶: Name of the artifact the archive is for.
cache: bool = False¶: Read the archive from the cache.

Returns¶

Archive of the requested artifact.

Return type¶