Record stores

zenpyre.record_stores ¶

Contain record stores.

zenpyre.record_stores.BaseRecordStore ¶

Bases: ABC

Abstract base class for record stores.

Defines the common interface that all record store implementations must provide. A concrete implementation would be, for example, :class:~zenpyre.record_stores.InMemoryRecordStore.

To implement a custom record store, subclass :class:BaseRecordStore and implement all abstract methods.

Implementations are expected to support use as a context manager (with SomeRecordStore(...) as store: ...), which calls :meth:close automatically on exit.

zenpyre.record_stores.BaseRecordStore.add_records `abstractmethod` ¶

add_records(records: list[Record]) -> None

Add or replace records in the store.

Records whose id already exists are replaced (upsert semantics). Since :class:~zenpyre.records.Record always has a required id, implementations do not need to validate its presence, though they may still choose to reject an empty string.

Parameters:

Name	Type	Description	Default
`records`	`list[Record]`	The list of :class:`~zenpyre.records.Record` instances to add.	required

zenpyre.record_stores.BaseRecordStore.all `abstractmethod` ¶

all() -> list[Record]

Return all records in the store.

Returns:

Type	Description
`list[Record]`	A list of all :class:`~zenpyre.records.Record` instances currently in the store.

zenpyre.record_stores.BaseRecordStore.check_ids `abstractmethod` ¶

check_ids(
    record_ids: list[str],
) -> tuple[list[str], list[str]]

Check which record IDs exist in the store.

Parameters:

Name	Type	Description	Default
`record_ids`	`list[str]`	The record IDs to check.	required

Returns:

Type	Description
`tuple[list[str], list[str]]`	A tuple of two lists: `(found, missing)` where `found` contains the IDs that exist in the store and `missing` contains the IDs that do not.

zenpyre.record_stores.BaseRecordStore.close `abstractmethod` ¶

close() -> None

Close the store and release any underlying resources (e.g. database connections, file handles).

Implementations should make repeated calls to close() safe (i.e. idempotent), since :meth:__exit__ calls it unconditionally and callers may also close a store manually before using it as a context manager.

zenpyre.record_stores.BaseRecordStore.count `abstractmethod` ¶

count() -> int

Return the total number of records in the store.

Returns:

Type	Description
`int`	The number of records currently stored.

zenpyre.record_stores.BaseRecordStore.delete `abstractmethod` ¶

delete(record_id: str) -> None

Delete a record by its ID.

IDs that do not exist should be silently ignored.

Parameters:

Name	Type	Description	Default
`record_id`	`str`	The ID of the record to delete.	required

zenpyre.record_stores.BaseRecordStore.delete_many `abstractmethod` ¶

delete_many(record_ids: list[str]) -> None

Delete multiple records by their IDs.

IDs that do not exist should be silently ignored.

Parameters:

Name	Type	Description	Default
`record_ids`	`list[str]`	The IDs of the records to delete.	required

zenpyre.record_stores.BaseRecordStore.filter `abstractmethod` ¶

filter(**metadata_filters: Any) -> list[Record]

Retrieve records matching all provided metadata filters.

All filters should be combined with AND. Each keyword argument matches the corresponding metadata key exactly.

Parameters:

Name	Type	Description	Default
`**metadata_filters`	`Any`	Key-value pairs where each key is a metadata field name and the value is the exact value to match. Calling with no arguments should return all records.	`{}`

Returns:

Type	Description
`list[Record]`	A list of matching :class:`~zenpyre.records.Record` instances.

zenpyre.record_stores.BaseRecordStore.get `abstractmethod` ¶

get(record_id: str) -> Record | None

Retrieve a single record by its ID.

Parameters:

Name	Type	Description	Default
`record_id`	`str`	The record ID to look up.	required

Returns:

Name	Type	Description
`The`	`Record \| None`	class:`~zenpyre.records.Record`, or `None` if not found.

zenpyre.record_stores.BaseRecordStore.get_many `abstractmethod` ¶

get_many(record_ids: list[str]) -> list[Record | None]

Retrieve multiple records by their IDs.

Parameters:

Name	Type	Description	Default
`record_ids`	`list[str]`	The record IDs to look up.	required

Returns:

Type	Description
`list[Record \| None]`	A list the same length as `record_ids`, with the corresponding :class:`~zenpyre.records.Record` for each ID that exists, or `None` for IDs not found.

zenpyre.record_stores.BaseRecordStore.iter_batches `abstractmethod` ¶

iter_batches(
    batch_size: int = 32,
) -> Generator[list[Record], None, None]

Yield records in batches, avoiding loading the whole store into memory at once.

This is the scalable equivalent of :meth:all: instead of materializing every record as a single list, it streams them from the database in chunks of batch_size.

Parameters:

Name	Type	Description	Default
`batch_size`	`int`	The maximum number of records to yield per batch. Must be a positive integer.	`32`

Yields:

Type	Description
`list[Record]`	Lists of records, each with at most `batch_size` elements, in the same order as :meth:`all`. The last batch may contain fewer than `batch_size` records.

Example

>>> from zenpyre.record_stores import InMemoryRecordStore
>>> from zenpyre.records import Record
>>> store = InMemoryRecordStore()
>>> store.add_records([Record(id=str(i), metadata={"index": i}) for i in range(5)])
>>> for batch in store.iter_batches(batch_size=2):
...     print(len(batch))
...
2
2
1

zenpyre.record_stores.BaseRecordStore.lazy_all ¶

lazy_all(batch_size: int = 32) -> Iterator[Record]

Lazily iterate over all records without loading them all into memory at once.

This is the streaming equivalent of :meth:all. The default implementation delegates to :meth:iter_batches and flattens the batches; implementations may override this with a more direct row-by-row cursor for a smaller memory footprint.

Parameters:

Name	Type	Description	Default
`batch_size`	`int`	The batch size used internally when pulling records from the underlying store. Does not affect the granularity of what is yielded -- records are always yielded one at a time.	`32`

Yields:

Name	Type	Description
`One`	`Record`	class:`~zenpyre.records.Record` at a time, in the same order as :meth:`all`.

Example

>>> from zenpyre.record_stores import InMemoryRecordStore
>>> from zenpyre.records import Record
>>> store = InMemoryRecordStore()
>>> store.add_records([Record(id=str(i), metadata={"index": i}) for i in range(3)])
>>> for record in store.lazy_all():
...     print(record.id)
...
0
1
2

zenpyre.record_stores.DuckDBRecordStore ¶

Bases: BaseDuckDBRecordStore

A DuckDB-backed store for :class:~zenpyre.records.Record objects.

Persists records to a DuckDB database and supports adding, retrieving, filtering, and deleting records. All metadata is stored as a JSON column, which provides flexibility for arbitrary metadata fields without requiring a fixed schema. For better query performance on known metadata fields, see :class:TypedDuckDBRecordStore.

Parameters:

Name	Type	Description	Default
`path`	`Path \| str`	Path to the DuckDB file, or `":memory:"` for an in-memory database (useful for testing).	`':memory:'`
`**kwargs`	`Any`	Additional keyword arguments to pass to `duckdb.connect`.	`{}`

Example

>>> from zenpyre.record_stores import DuckDBRecordStore
>>> from zenpyre.records import Record
>>> store = DuckDBRecordStore(":memory:")
>>> records = [
...     Record(
...         id="1",
...         metadata={"author": "Alice", "category": "Programming"},
...     ),
...     Record(
...         id="2",
...         metadata={"author": "Alice", "category": "Programming"},
...     ),
...     Record(
...         id="3",
...         metadata={"author": "Bob", "category": "History"},
...     ),
... ]
>>> store.add_records(records)
>>> len(store.filter(author="Alice"))
2
>>> len(store.filter(author="Alice", category="Programming"))
2
>>> len(store.filter(category="History"))
1

zenpyre.record_stores.InMemoryRecordStore ¶

Bases: BaseRecordStore, InlineDisplayMixin

A :class:~zenpyre.record_stores.base.BaseRecordStore implementation backed by a plain dict.

Records are keyed by their id and held entirely in process memory -- nothing is persisted to disk. This is primarily useful for testing, small-scale exploration, or pipelines that don't need durability.

Records are deep-copied on both write and read so that mutating a :class:~zenpyre.records.Record returned by this store (or a record passed into :meth:add_records) never affects the store's internal state. This still matters even though Record is a frozen dataclass: frozen=True only prevents reassigning the metadata attribute itself, not mutating the dict it points to in place (e.g. record.metadata["key"] = "value" still works). This trades some performance for isolation; for very large metadata payloads or hot loops, consider a store that doesn't copy on every access.

Example

>>> from zenpyre.records import Record
>>> from zenpyre.record_stores import InMemoryRecordStore
>>> store = InMemoryRecordStore()
>>> store.add_records([Record(id="1", metadata={"source": "hello"})])
>>> store.count()
1
>>> store.get("1").metadata
{'source': 'hello'}

zenpyre.record_stores.SQLiteRecordStore ¶

Bases: BaseSQLiteRecordStore

A SQLite-backed store for :class:~zenpyre.records.Record objects.

Persists records to a SQLite database and supports adding, retrieving, filtering, and deleting records. All metadata is stored as a JSON column (using SQLite's built-in json1 functions), which provides flexibility for arbitrary metadata fields without requiring a fixed schema.

The constructor mirrors :func:sqlite3.connect directly. For the common case of opening a file by path (optionally read-only), use :meth:from_path instead.

Parameters:

Name	Type	Description	Default
`database`	`Path \| str`	The `database` argument passed to `sqlite3.connect` (path, `":memory:"`, or `file:` URI).	`':memory:'`
`**kwargs`	`Any`	Additional keyword arguments to pass to `sqlite3.connect`.	`{}`

Example

>>> from zenpyre.record_stores import SQLiteRecordStore
>>> from zenpyre.records import Record
>>> store = SQLiteRecordStore(":memory:")
>>> records = [
...     Record(
...         id="1",
...         metadata={"author": "Alice", "category": "Programming"},
...     ),
...     Record(
...         id="2",
...         metadata={"author": "Alice", "category": "Programming"},
...     ),
...     Record(
...         id="3",
...         metadata={"author": "Bob", "category": "History"},
...     ),
... ]
>>> store.add_records(records)
>>> len(store.filter(author="Alice"))
2
>>> len(store.filter(author="Alice", category="Programming"))
2
>>> len(store.filter(category="History"))
1

zenpyre.record_stores.SQLiteRecordStore.from_path `classmethod` ¶

from_path(
    path: Path | str,
    *,
    read_only: bool = False,
    **kwargs: Any
) -> SQLiteRecordStore

Construct a :class:SQLiteRecordStore from a file path.

Builds the appropriate file: URI for sqlite3.connect, including read-only access, so callers don't need to construct SQLite URIs themselves.

Parameters:

Name	Type	Description	Default
`path`	`Path \| str`	Path to the SQLite file, or `":memory:"` for an in-memory database (useful for testing).	required
`read_only`	`bool`	If `True`, open the database in read-only mode. The database file must already exist.	`False`
`**kwargs`	`Any`	Additional keyword arguments to pass to `sqlite3.connect`.	`{}`

Returns:

Type	Description
`SQLiteRecordStore`	A new :class:`SQLiteRecordStore` connected to `path`.

zenpyre.record_stores.TypedDuckDBRecordStore ¶

Bases: BaseDuckDBRecordStore

A DuckDB-backed store for records with metadata filtering.

Persists records to a DuckDB database and supports adding, retrieving, and filtering by metadata fields. An optional metadata_schema maps known metadata field names to DuckDB types. Known fields are stored as typed columns for fast, index-friendly queries. Any metadata fields not in the schema are stored in an extra JSON overflow column, so nothing is lost.

Parameters:

Name	Type	Description	Default
`path`	`Path \| str`	Path to the DuckDB file, or `":memory:"` for an in-memory database (useful for testing).	`':memory:'`
`metadata_schema`	`dict[str, str] \| None`	Optional mapping of metadata field names to DuckDB type strings (e.g. `{"ticker": "VARCHAR", "cik": "INTEGER", "year": "INTEGER"}`). Fields in the schema get native typed columns; all other metadata fields go into the `extra` JSON overflow column. Defaults to `None`, which stores all metadata as JSON only.	`None`
`**kwargs`	`Any`	Additional keyword arguments to pass to `duckdb.connect`.	`{}`

Example

>>> from zenpyre.record_stores import TypedDuckDBRecordStore
>>> from zenpyre.records import Record
>>> schema = {"author": "VARCHAR", "year": "INTEGER", "category": "VARCHAR"}
>>> store = TypedDuckDBRecordStore(":memory:", metadata_schema=schema)
>>> records = [
...     Record(
...         id="1",
...         metadata={"author": "Alice", "year": 2022, "category": "Programming"},
...     ),
...     Record(
...         id="2",
...         metadata={"author": "Alice", "year": 2023, "category": "Programming"},
...     ),
...     Record(
...         id="3",
...         metadata={"author": "Bob", "year": 2021, "category": "History"},
...     ),
... ]
>>> store.add_records(records)
>>> len(store.filter(author="Alice"))
2
>>> len(store.filter(author="Alice", category="Programming"))
2
>>> len(store.filter(category="History"))
1

zenpyre.record_stores.TypedSQLiteRecordStore ¶

Bases: BaseSQLiteRecordStore

A SQLite-backed store for records with metadata filtering.

Persists records to a SQLite database and supports adding, retrieving, and filtering by metadata fields. An optional metadata_schema maps known metadata field names to SQLite types. Known fields are stored as typed columns for fast, index-friendly queries. Any metadata fields not in the schema are stored in an extra JSON overflow column, so nothing is lost.

The constructor mirrors :func:sqlite3.connect directly (plus the metadata_schema argument). For the common case of opening a file by path (optionally read-only), use :meth:from_path instead.

Parameters:

Name	Type	Description	Default
`database`	`Path \| str`	The `database` argument passed to `sqlite3.connect` (path, `":memory:"`, or `file:` URI).	`':memory:'`
`metadata_schema`	`dict[str, str] \| None`	Optional mapping of metadata field names to SQLite type strings (e.g. `{"ticker": "TEXT", "cik": "INTEGER", "year": "INTEGER"}`). Fields in the schema get native typed columns; all other metadata fields go into the `extra` JSON overflow column. Defaults to `None`, which stores all metadata as JSON only.	`None`
`**kwargs`	`Any`	Additional keyword arguments to pass to `sqlite3.connect`.	`{}`

Example

>>> from zenpyre.record_stores import TypedSQLiteRecordStore
>>> from zenpyre.records import Record
>>> schema = {"author": "TEXT", "year": "INTEGER", "category": "TEXT"}
>>> store = TypedSQLiteRecordStore(":memory:", metadata_schema=schema)
>>> records = [
...     Record(
...         id="1",
...         metadata={"author": "Alice", "year": 2022, "category": "Programming"},
...     ),
...     Record(
...         id="2",
...         metadata={"author": "Alice", "year": 2023, "category": "Programming"},
...     ),
...     Record(
...         id="3",
...         metadata={"author": "Bob", "year": 2021, "category": "History"},
...     ),
... ]
>>> store.add_records(records)
>>> len(store.filter(author="Alice"))
2
>>> len(store.filter(author="Alice", category="Programming"))
2
>>> len(store.filter(category="History"))
1

zenpyre.record_stores.TypedSQLiteRecordStore.from_path `classmethod` ¶

from_path(
    path: Path | str,
    *,
    metadata_schema: dict[str, str] | None = None,
    read_only: bool = False,
    **kwargs: Any
) -> TypedSQLiteRecordStore

Construct a :class:TypedSQLiteRecordStore from a file path.

Builds the appropriate file: URI for sqlite3.connect, including read-only access, so callers don't need to construct SQLite URIs themselves.

Parameters:

Name	Type	Description	Default
`path`	`Path \| str`	Path to the SQLite file, or `":memory:"` for an in-memory database (useful for testing).	required
`metadata_schema`	`dict[str, str] \| None`	Optional mapping of metadata field names to SQLite type strings. See the class docstring.	`None`
`read_only`	`bool`	If `True`, open the database in read-only mode. The database file must already exist.	`False`
`**kwargs`	`Any`	Additional keyword arguments to pass to `sqlite3.connect`.	`{}`

Returns:

Type	Description
`TypedSQLiteRecordStore`	A new :class:`TypedSQLiteRecordStore` connected to `path`.

zenpyre.record_stores.resolve_record_store ¶

resolve_record_store(
    record_store: (
        BaseRecordStore | dict[str, Any] | BaseConfig
    ),
) -> BaseRecordStore

Resolve a :class:~zenpyre.record_stores.base.BaseRecordStore instance from an existing object, a configuration dictionary, or a :class:~zenpyre.utils.config.BaseConfig.

If record_store is already a :class:~zenpyre.record_stores.base.BaseRecordStore instance it is returned as-is. If it is a :class:dict or a :class:~zenpyre.utils.config.BaseConfig, it is treated as an objectory factory configuration and instantiated via :func:objectory.factory. See :func:~zenpyre.utils.resolve.resolve_object for details.

Parameters:

Name	Type	Description	Default
`record_store`	`BaseRecordStore \| dict[str, Any] \| BaseConfig`	Either a fully configured :class:`~zenpyre.record_stores.base.BaseRecordStore` instance, a :class:`dict` containing an `objectory` factory specification (must include a `"_target_"` key pointing to the fully-qualified class name), or a :class:`~zenpyre.utils.config.BaseConfig` whose `to_kwargs()` includes a `"_target_"` key.	required

Returns:

Type	Description
`BaseRecordStore`	A configured
`BaseRecordStore`	class:`~zenpyre.record_stores.base.BaseRecordStore`
`BaseRecordStore`	instance.

Raises:

Type	Description
`TypeError`	If the resolved object is not a :class:`~zenpyre.record_stores.base.BaseRecordStore` instance.

Example

>>> from zenpyre.record_stores import InMemoryRecordStore, resolve_record_store
>>> # From an existing instance:
>>> record_store = resolve_record_store(InMemoryRecordStore())
>>> # From a configuration dictionary:
>>> record_store = resolve_record_store(
...     {"_target_": "zenpyre.record_stores.InMemoryRecordStore"}
... )

zenpyre.record_stores.factory ¶

Contain factories for record stores.

zenpyre.record_stores.factory.BaseRecordStoreFactory ¶

Bases: ABC

Abstract base class for :class:~zenpyre.record_stores.base.BaseRecordStore factories.

Subclasses implement :meth:make_record_store to instantiate and return a configured :class:~zenpyre.record_stores.base.BaseRecordStore object. This pattern decouples record store creation from the rest of the codebase, making it easy to swap record stores (e.g. in-memory, SQLite, DuckDB) without changing call sites.

Example

>>> from zenpyre.record_stores import InMemoryRecordStore
>>> from zenpyre.record_stores.base import BaseRecordStore
>>> from zenpyre.record_stores.factory import BaseRecordStoreFactory
>>> class MyRecordStoreFactory(BaseRecordStoreFactory):
...     def make_record_store(self) -> BaseRecordStore:
...         return InMemoryRecordStore()
...
>>> factory = MyRecordStoreFactory()
>>> record_store = factory.make_record_store()

zenpyre.record_stores.factory.BaseRecordStoreFactory.make_record_store `abstractmethod` ¶

make_record_store() -> BaseRecordStore

Create and return a configured BaseRecordStore instance.

Returns:

Name	Type	Description
`A`	`BaseRecordStore`	class:`~zenpyre.record_stores.base.BaseRecordStore`
	`BaseRecordStore`	instance ready for use.

zenpyre.record_stores.factory.ConfigurableRecordStoreFactory ¶

Bases: BaseRecordStoreFactory, MultilineDisplayMixin

A concrete BaseRecordStore factory that accepts either a pre- built :class:~zenpyre.record_stores.base.BaseRecordStore instance or a configuration dictionary.

When a dict is provided it is resolved at each :meth:make_record_store call via :func:~zenpyre.record_stores.resolve.resolve_record_store, which uses objectory to instantiate the configured class. When an instance is provided it is returned as-is.

Parameters:

Name	Type	Description	Default
`record_store`	`BaseRecordStore \| dict[str, Any]`	A fully configured :class:`~zenpyre.record_stores.base.BaseRecordStore` instance, or a :class:`dict` containing an `objectory` factory specification (must include a `"_target_"` key pointing to the fully-qualified class name).	required

Example

>>> from zenpyre.record_stores import InMemoryRecordStore
>>> from zenpyre.record_stores.factory import ConfigurableRecordStoreFactory
>>> factory = ConfigurableRecordStoreFactory(InMemoryRecordStore())
>>> record_store = factory.make_record_store()

zenpyre.record_stores.factory.DuckDBRecordStoreFactory ¶

Bases: BaseRecordStoreFactory, MultilineDisplayMixin

A concrete BaseRecordStore factory that builds a :class:~zenpyre.record_stores.DuckDBRecordStore backed by a DuckDB file at a given path.

Use this when you want a factory that lazily constructs a fresh :class:~zenpyre.record_stores.DuckDBRecordStore at path each time :meth:make_record_store is called, rather than wrapping an already-instantiated store.

Parameters:

Name	Type	Description	Default
`path`	`Path \| str`	The path to the DuckDB file used to back the record store.	required
`**kwargs`	`Any`	Additional keyword arguments forwarded to :class:`~zenpyre.record_stores.DuckDBRecordStore`.	`{}`

Example

>>> from pathlib import Path
>>> from zenpyre.record_stores.factory import DuckDBRecordStoreFactory
>>> factory = DuckDBRecordStoreFactory(Path("/tmp/my_app/records.duckdb"))
>>> record_store = factory.make_record_store()  # doctest: +SKIP

zenpyre.record_stores.factory.InMemoryRecordStoreFactory ¶

Bases: BaseRecordStoreFactory, MultilineDisplayMixin

A concrete BaseRecordStore factory that builds a fresh :class:~zenpyre.record_stores.InMemoryRecordStore on each :meth:make_record_store call.

Use this when you want a factory that lazily constructs a new, empty :class:~zenpyre.record_stores.InMemoryRecordStore each time :meth:make_record_store is called, rather than wrapping an already-instantiated store (see :class:~zenpyre.record_stores.factory.RecordStoreFactory for that).

Example

>>> from zenpyre.record_stores.factory import InMemoryRecordStoreFactory
>>> factory = InMemoryRecordStoreFactory()
>>> record_store = factory.make_record_store()

zenpyre.record_stores.factory.RecordStoreFactory ¶

Bases: BaseRecordStoreFactory, MultilineDisplayMixin

A concrete BaseRecordStore factory that wraps a pre-built :class:~zenpyre.record_stores.base.BaseRecordStore instance.

Use this when the record store is already instantiated and you simply want to wrap it in the :class:~BaseRecordStoreFactory interface — for example, when injecting a fixed record store into a component that expects a factory.

Parameters:

Name	Type	Description	Default
`record_store`	`BaseRecordStore`	A fully configured :class:`~zenpyre.record_stores.base.BaseRecordStore` instance to return from :meth:`make_record_store`.	required

Example

>>> from zenpyre.record_stores import InMemoryRecordStore
>>> from zenpyre.record_stores.factory import RecordStoreFactory
>>> factory = RecordStoreFactory(InMemoryRecordStore())
>>> record_store = factory.make_record_store()

zenpyre.record_stores.factory.SQLiteRecordStoreFactory ¶

Bases: BaseRecordStoreFactory, MultilineDisplayMixin

A concrete BaseRecordStore factory that builds a :class:~zenpyre.record_stores.SQLiteRecordStore backed by a SQLite file at a given path.

Use this when you want a factory that lazily constructs a fresh :class:~zenpyre.record_stores.SQLiteRecordStore at path (via :meth:~zenpyre.record_stores.SQLiteRecordStore.from_path) each time :meth:make_record_store is called, rather than wrapping an already-instantiated store.

Parameters:

Name	Type	Description	Default
`path`	`Path \| str`	The path to the SQLite file used to back the record store, or `":memory:"` for an in-memory database.	required
`read_only`	`bool`	If `True`, open the database in read-only mode. The database file must already exist.	`False`
`**kwargs`	`Any`	Additional keyword arguments forwarded to :meth:`~zenpyre.record_stores.SQLiteRecordStore.from_path`.	`{}`

Example

>>> from pathlib import Path
>>> from zenpyre.record_stores.factory import SQLiteRecordStoreFactory
>>> factory = SQLiteRecordStoreFactory(Path("/tmp/my_app/records.sqlite"))
>>> record_store = factory.make_record_store()  # doctest: +SKIP

zenpyre.record_stores.factory.TypedDuckDBRecordStoreFactory ¶

Bases: BaseRecordStoreFactory, MultilineDisplayMixin

A concrete BaseRecordStore factory that builds a :class:~zenpyre.record_stores.TypedDuckDBRecordStore backed by a DuckDB file at a given path.

Use this when you want a factory that lazily constructs a fresh :class:~zenpyre.record_stores.TypedDuckDBRecordStore at path each time :meth:make_record_store is called, rather than wrapping an already-instantiated store.

Parameters:

Name	Type	Description	Default
`path`	`Path \| str`	The path to the DuckDB file used to back the record store.	required
`metadata_schema`	`dict[str, str] \| None`	Optional mapping of metadata field names to DuckDB type strings. See :class:`~zenpyre.record_stores.TypedDuckDBRecordStore`'s docstring for details.	`None`
`**kwargs`	`Any`	Additional keyword arguments forwarded to :class:`~zenpyre.record_stores.TypedDuckDBRecordStore`.	`{}`

Example

>>> from pathlib import Path
>>> from zenpyre.record_stores.factory import TypedDuckDBRecordStoreFactory
>>> factory = TypedDuckDBRecordStoreFactory(Path("/tmp/my_app/records.duckdb"))
>>> record_store = factory.make_record_store()  # doctest: +SKIP

zenpyre.record_stores.factory.TypedSQLiteRecordStoreFactory ¶

Bases: BaseRecordStoreFactory, MultilineDisplayMixin

A concrete BaseRecordStore factory that builds a :class:~zenpyre.record_stores.TypedSQLiteRecordStore backed by a SQLite file at a given path.

Use this when you want a factory that lazily constructs a fresh :class:~zenpyre.record_stores.TypedSQLiteRecordStore at path (via :meth:~zenpyre.record_stores.TypedSQLiteRecordStore.from_path) each time :meth:make_record_store is called, rather than wrapping an already-instantiated store.

Parameters:

Name	Type	Description	Default
`path`	`Path \| str`	The path to the SQLite file used to back the record store, or `":memory:"` for an in-memory database.	required
`metadata_schema`	`dict[str, str] \| None`	Optional mapping of metadata field names to SQLite type strings. See :class:`~zenpyre.record_stores.TypedSQLiteRecordStore`'s docstring for details.	`None`
`read_only`	`bool`	If `True`, open the database in read-only mode. The database file must already exist.	`False`
`**kwargs`	`Any`	Additional keyword arguments forwarded to :meth:`~zenpyre.record_stores.TypedSQLiteRecordStore.from_path`.	`{}`

Example

>>> from pathlib import Path
>>> from zenpyre.record_stores.factory import TypedSQLiteRecordStoreFactory
>>> factory = TypedSQLiteRecordStoreFactory(Path("/tmp/my_app/records.sqlite"))
>>> record_store = factory.make_record_store()  # doctest: +SKIP

Record stores

zenpyre.record_stores ¶

zenpyre.record_stores.BaseRecordStore ¶

zenpyre.record_stores.BaseRecordStore.add_records abstractmethod ¶

zenpyre.record_stores.BaseRecordStore.all abstractmethod ¶

zenpyre.record_stores.BaseRecordStore.check_ids abstractmethod ¶

zenpyre.record_stores.BaseRecordStore.close abstractmethod ¶

zenpyre.record_stores.BaseRecordStore.count abstractmethod ¶

zenpyre.record_stores.BaseRecordStore.delete abstractmethod ¶

zenpyre.record_stores.BaseRecordStore.delete_many abstractmethod ¶

zenpyre.record_stores.BaseRecordStore.filter abstractmethod ¶

zenpyre.record_stores.BaseRecordStore.get abstractmethod ¶

zenpyre.record_stores.BaseRecordStore.get_many abstractmethod ¶

zenpyre.record_stores.BaseRecordStore.iter_batches abstractmethod ¶

zenpyre.record_stores.BaseRecordStore.lazy_all ¶

zenpyre.record_stores.DuckDBRecordStore ¶

zenpyre.record_stores.InMemoryRecordStore ¶

zenpyre.record_stores.SQLiteRecordStore ¶

zenpyre.record_stores.SQLiteRecordStore.from_path classmethod ¶

zenpyre.record_stores.TypedDuckDBRecordStore ¶

zenpyre.record_stores.TypedSQLiteRecordStore ¶

zenpyre.record_stores.TypedSQLiteRecordStore.from_path classmethod ¶

zenpyre.record_stores.resolve_record_store ¶

zenpyre.record_stores.factory ¶

zenpyre.record_stores.factory.BaseRecordStoreFactory ¶

zenpyre.record_stores.factory.BaseRecordStoreFactory.make_record_store abstractmethod ¶

zenpyre.record_stores.factory.ConfigurableRecordStoreFactory ¶

zenpyre.record_stores.factory.DuckDBRecordStoreFactory ¶

zenpyre.record_stores.factory.InMemoryRecordStoreFactory ¶

zenpyre.record_stores.factory.RecordStoreFactory ¶

zenpyre.record_stores.factory.SQLiteRecordStoreFactory ¶

zenpyre.record_stores.factory.TypedDuckDBRecordStoreFactory ¶

zenpyre.record_stores.factory.TypedSQLiteRecordStoreFactory ¶

zenpyre.record_stores.BaseRecordStore.add_records `abstractmethod` ¶

zenpyre.record_stores.BaseRecordStore.all `abstractmethod` ¶

zenpyre.record_stores.BaseRecordStore.check_ids `abstractmethod` ¶

zenpyre.record_stores.BaseRecordStore.close `abstractmethod` ¶

zenpyre.record_stores.BaseRecordStore.count `abstractmethod` ¶

zenpyre.record_stores.BaseRecordStore.delete `abstractmethod` ¶

zenpyre.record_stores.BaseRecordStore.delete_many `abstractmethod` ¶

zenpyre.record_stores.BaseRecordStore.filter `abstractmethod` ¶

zenpyre.record_stores.BaseRecordStore.get `abstractmethod` ¶

zenpyre.record_stores.BaseRecordStore.get_many `abstractmethod` ¶

zenpyre.record_stores.BaseRecordStore.iter_batches `abstractmethod` ¶

zenpyre.record_stores.SQLiteRecordStore.from_path `classmethod` ¶

zenpyre.record_stores.TypedSQLiteRecordStore.from_path `classmethod` ¶

zenpyre.record_stores.factory.BaseRecordStoreFactory.make_record_store `abstractmethod` ¶