Record

persista.record

Define the Record container and helpers to filter and sort collections of records by their metadata.

persista.record.Record dataclass

A generic immutable record with a stable UUID identifier and arbitrary metadata.

Use :meth:from_metadata as the preferred constructor to automatically derive a stable UUID from the metadata dict.

Note

metadata is copied into a read-only :class:~types.MappingProxyType view during construction, so record.metadata["x"] = 1 raises TypeError instead of silently mutating the record - the dict passed in to metadata is not itself frozen, so mutating it directly after construction still does not affect the record. Instances remain unhashable, since a MappingProxyType is itself unhashable and eq=True (the dataclass default) disables the auto-generated __hash__. Use record.id as the hashable identifier instead.

Parameters:

Name	Type	Description	Default
id	str	Unique identifier for the record, typically a UUID derived from the metadata via :func:~coola.identifier.generate_stable_uuid5.	required
metadata	dict[str, Any]	Arbitrary key-value metadata associated with the record. Defaults to an empty dict. Copied into a read-only view; the original dict is left untouched.	dict()

persista.record.Record.from_metadata classmethod

from_metadata(metadata: dict[str, Any]) -> Record

Construct a :class:Record from a metadata dict.

Computes a stable UUID from metadata via :func:~coola.identifier.generate_stable_uuid5 and uses it as the record's id. Two calls with the same metadata contents (regardless of key insertion order) will produce the same id.

Parameters:

Name	Type	Description	Default
metadata	dict[str, Any]	Arbitrary key-value metadata to associate with the record. Used to derive the id.	required

Returns:

Type	Description
Record	A new :class:Record with id derived from
Record	metadata.

Example

>>> from persista.record import Record
>>> record = Record.from_metadata({"source": "cats.txt", "page": 1})
>>> record.id  # doctest: +ELLIPSIS
'...'
>>> dict(record.metadata)
{'source': 'cats.txt', 'page': 1}

persista.record.filter_by_metadata

filter_by_metadata(
    records: list[Record], metadata_key: str, value: Any
) -> list[Record]

Filter a list of records by the value of a metadata key.

Returns a new list containing only records whose metadata contains metadata_key with a value equal to value. Records missing metadata_key are excluded, even when value=None: a record missing metadata_key never matches, regardless of value.

Parameters:

Name	Type	Description	Default
records	list[Record]	The list of :class:~persista.record.Record instances to filter.	required
metadata_key	str	The metadata key to filter by.	required
value	Any	The value to match against. Records whose metadata_key equals this value are kept.	required

Returns:

Type	Description
list[Record]	A new list of :class:~persista.record.Record instances whose metadata
list[Record]	matches the filter. The original list is not modified.

Example

>>> from persista.record import Record, filter_by_metadata
>>> records = [
...     Record(id="a", metadata={"category": "Science"}),
...     Record(id="b", metadata={"category": "Cooking"}),
...     Record(id="c", metadata={"category": "Science"}),
... ]
>>> result = filter_by_metadata(records, "category", "Science")
>>> [r.id for r in result]
['a', 'c']

persista.record.filter_by_metadata_range

filter_by_metadata_range(
    records: list[Record],
    metadata_key: str,
    lower: Any = None,
    upper: Any = None,
) -> list[Record]

Filter a list of records by a range of values for a metadata key.

Returns a new list containing only records whose metadata contains metadata_key with a value within the specified range [lower, upper] (inclusive on both ends). Either bound can be set to None to indicate no constraint on that side. If both bounds are None, all records that contain metadata_key are returned. Records missing metadata_key are always excluded.

Parameters:

Name	Type	Description	Default
records	list[Record]	The list of :class:~persista.record.Record instances to filter.	required
metadata_key	str	The metadata key to filter by.	required
lower	Any	The inclusive lower bound. Pass None (the default) for no lower bound.	None
upper	Any	The inclusive upper bound. Pass None (the default) for no upper bound.	None

Returns:

Type	Description
list[Record]	A new list of :class:~persista.record.Record instances whose
list[Record]	metadata_key value falls within [lower, upper]. The
list[Record]	original list is not modified.

Raises:

Type	Description
TypeError	If the metadata values are not comparable with the provided bounds (e.g. comparing str with int).

Example

>>> from persista.record import Record, filter_by_metadata_range
>>> records = [
...     Record(id="a", metadata={"page": 1}),
...     Record(id="b", metadata={"page": 5}),
...     Record(id="c", metadata={"page": 10}),
... ]
>>> result = filter_by_metadata_range(records, "page", lower=2, upper=8)
>>> [r.id for r in result]
['b']
>>> result = filter_by_metadata_range(records, "page", lower=5)
>>> [r.id for r in result]
['b', 'c']
>>> result = filter_by_metadata_range(records, "page", upper=5)
>>> [r.id for r in result]
['a', 'b']

persista.record.filter_by_metadata_values

filter_by_metadata_values(
    records: list[Record],
    metadata_key: str,
    values: set[Any],
) -> list[Record]

Filter a list of records by checking if a metadata value is in a set.

Returns a new list containing only records whose metadata contains metadata_key with a value that is a member of values. Records missing metadata_key are excluded, even when None is included in values: a record missing metadata_key never matches.

Parameters:

Name	Type	Description	Default
records	list[Record]	The list of :class:~persista.record.Record instances to filter.	required
metadata_key	str	The metadata key to filter by.	required
values	set[Any]	The set of accepted values. Records whose metadata_key is in this set are kept.	required

Returns:

Type	Description
list[Record]	A new list of :class:~persista.record.Record instances whose
list[Record]	metadata_key value is in values. The original list is
list[Record]	not modified.

Example

>>> from persista.record import Record, filter_by_metadata_values
>>> records = [
...     Record(id="a", metadata={"category": "Science"}),
...     Record(id="b", metadata={"category": "Cooking"}),
...     Record(id="c", metadata={"category": "Technology"}),
...     Record(id="d", metadata={"category": "Science"}),
... ]
>>> result = filter_by_metadata_values(records, "category", {"Science", "Technology"})
>>> sorted(r.id for r in result)
['a', 'c', 'd']

persista.record.sort_by_metadata

sort_by_metadata(
    records: list[Record],
    metadata_key: str,
    *,
    keep_missing: bool = True,
    reverse: bool = False,
) -> list[Record]

Sort a list of records by the value of a metadata key.

Records are sorted in ascending order by the value of metadata_key by default, or descending order if reverse=True. Records that do not contain metadata_key in their metadata are placed at the end of the result by default, or removed entirely if keep_missing=False.

Parameters:

Name	Type	Description	Default
records	list[Record]	The list of :class:~persista.record.Record instances to sort.	required
metadata_key	str	The metadata key to sort by.	required
keep_missing	bool	If True (the default), records without metadata_key in their metadata are kept and placed at the end of the result. If False, they are excluded from the result entirely.	True
reverse	bool	If True, the result is sorted in descending order. Defaults to False, matching the behaviour of :func:sorted.	False

Returns:

Type	Description
list[Record]	A new sorted list of :class:~persista.record.Record instances. The
list[Record]	original list is not modified.

Raises:

Type	Description
TypeError	If the metadata values for metadata_key are not mutually comparable (e.g. mixing str and int).

Example

>>> from persista.record import Record, sort_by_metadata
>>> records = [
...     Record(id="b", metadata={"source": "b.txt"}),
...     Record(id="a", metadata={"source": "a.txt"}),
...     Record(id="c"),
... ]
>>> sorted_records = sort_by_metadata(records, "source")
>>> [r.metadata.get("source") for r in sorted_records]
['a.txt', 'b.txt', None]
>>> sorted_records = sort_by_metadata(records, "source", reverse=True)
>>> [r.metadata.get("source") for r in sorted_records]
['b.txt', 'a.txt', None]
>>> sorted_records = sort_by_metadata(records, "source", keep_missing=False)
>>> [r.metadata.get("source") for r in sorted_records]
['a.txt', 'b.txt']

persista.record.analysis

Contain code to analyze records.

persista.record.analysis.MetadataStats dataclass

Streaming record metadata health/stats.

Accumulates statistics one record at a time via update, so it can be fed from a list, a generator, or any other iterable without requiring the full corpus to be held in memory at once. Call to_dict at the end to compute the final report.

Note

Once a key's sample of unique values has reached n_sample_values, no further values for that key are added to the sample - the sample simply reflects the first n_sample_values distinct values encountered for that key, in record order. This makes the sample deterministic (independent of set iteration order) at the cost of not necessarily being a uniform sample over the whole corpus.

Memory

• O(number of distinct metadata keys) for the per-key counters.
• O(number of distinct keys x n_sample_values) for the sampled unique values, bounded by n_sample_values per key regardless of corpus size. If n_sample_values is None, this becomes O(number of distinct keys x number of distinct values per key) instead, which can grow unbounded for high-cardinality keys (e.g. UUIDs).

With a numeric n_sample_values, this is effectively O(1) with respect to the number of records, so it scales to corpora too large to fit in memory.

Attributes:

Name	Type	Description
n_sample_values	int | None	Max number of unique sample values retained per metadata key, or None to retain all unique values for every key (no cap).
count	int	Total number of records processed.
missing_metadata_count	int	Number of records whose metadata is empty or falsy.
total_keys	int	Sum of the number of metadata keys across all records processed.
min_keys	int | None	Number of metadata keys on the record with the fewest keys seen so far, or None if no records have been processed yet.
max_keys	int | None	Number of metadata keys on the record with the most keys seen so far, or None if no records have been processed yet.

persista.record.analysis.MetadataStats.to_dict

to_dict() -> dict[str, Any]

Compute the final statistics report from the accumulated state.

Safe to call multiple times and at any point during accumulation; it does not mutate any state and always reflects everything processed via update so far.

Returns:

Type	Description
dict[str, Any]	A dict with the following keys:
dict[str, Any]	count: Total number of records processed.
dict[str, Any]	missing_metadata_count: Number of records with empty/missing metadata.
dict[str, Any]	avg_keys: Mean number of metadata keys per record, or 0 if no records were processed.
dict[str, Any]	min_keys / max_keys: Fewest/most metadata keys seen on any single record, or None if empty.
dict[str, Any]	distinct_keys_seen: Number of distinct metadata keys observed across all records.
dict[str, Any]	per_key: A dict mapping each metadata key to a dict with: present_in_records: Number of records containing the key. missing_in_records: Number of records not containing the key. value_types: Sorted list of type(value).__name__ seen for the key. none_or_empty_count: Number of records where the key's value is None or the empty string. unique_values_sample: Sorted sample of unique values seen for the key, capped at n_sample_values entries (or all of them, if n_sample_values is None). Reflects the first distinct values encountered, in record order. unique_values_sample_truncated: True if not all unique values for the key could be retained (either because the sample cap was hit or because an unhashable value was encountered). Always False when n_sample_values is None, unless an unhashable value was seen.
dict[str, Any]	Percentages are intentionally omitted - compute them later
dict[str, Any]	from the raw counts (e.g. present_in_records / count)
dict[str, Any]	since they are trivially derived from this report.

persista.record.analysis.MetadataStats.update

update(record: Record) -> None

Fold a single record's metadata into the running statistics.

Updates the record-level counters (count, missing metadata, keys-per-record min/max/total) as well as the per-key counters, value-type sets, none/empty counts, and unique-value samples.

Parameters:

Name	Type	Description	Default
record	Record	The record to process. Its metadata is expected to be a mapping; a falsy/empty mapping is counted via missing_metadata_count and contributes zero keys.	required

persista.record.analysis.compute_metadata_stats

compute_metadata_stats(
    records: Iterable[Record],
    n_sample_values: int | None = 3,
) -> dict[str, Any]

Compute metadata health/statistics over a stream of records.

Streaming health/stat check over a list, generator, or any iterable of Records. Records are consumed one at a time, so this works with iterators or generators whose full contents cannot fit in memory — only per-key aggregates are retained (see RecordMetadataStats for the memory profile).

Parameters:

Name	Type	Description	Default
records	Iterable[Record]	A list, generator, or other iterable of zenpyre.records.Record objects. Consumed exactly once; if a generator/iterator is passed in, it will be exhausted by this call.	required
n_sample_values	int | None	Max number of unique sample values to retain per metadata key. Defaults to 3. Pass None to track every unique value for every key instead of a bounded sample - useful for smaller corpora or low-cardinality metadata, but memory usage is then unbounded with respect to the number of distinct values per key.	3

Returns:

Type	Description
dict[str, Any]	A dict of statistics as described in
dict[str, Any]	RecordMetadataStats.to_dict. For an empty input, returns a
dict[str, Any]	report with count of 0 and None/0/empty values
dict[str, Any]	for the remaining fields.

Example

>>> from persista.record import Record
>>> from persista.record.analysis import compute_metadata_stats
>>> records = [
...     Record(id="a", metadata={"source": "a.pdf"}),
...     Record(id="b", metadata={"source": "b.pdf", "page": 1}),
... ]
>>> stats = compute_metadata_stats(records)
>>> stats["count"]
2

persista.record.analysis.compute_schema_shapes

compute_schema_shapes(
    records: Iterable[Record],
) -> dict[tuple[str, ...], list[Any]]

Group record ids by the shape of their metadata, i.e. the set of keys present, ignoring the keys' values.

This complements compute_metadata_stats, which reports per-key statistics across the whole corpus but does not show which combinations of keys co-occur. A corpus with schema drift (e.g. some records carrying an extra field added later, or missing one dropped later) shows up here as multiple distinct shapes.

Records are consumed one at a time, so this works with generators or other iterables whose full contents cannot fit in memory. Memory usage is O(number of distinct shapes), which is typically small even for large corpora since most pipelines only produce a handful of distinct metadata schemas.

Parameters:

Name	Type	Description	Default
records	Iterable[Record]	A list, generator, or other iterable of persista.record.Record objects. Consumed exactly once; if a generator/iterator is passed in, it will be exhausted by this call.	required

Returns:

Type	Description
dict[tuple[str, ...], list[Any]]	A dict mapping each distinct shape (a tuple of metadata keys,
dict[tuple[str, ...], list[Any]]	sorted alphabetically) to the list of record ids having
dict[tuple[str, ...], list[Any]]	that shape, in the order their records were encountered. Keys
dict[tuple[str, ...], list[Any]]	are in the order their shape was first encountered. A record
dict[tuple[str, ...], list[Any]]	with no metadata has the shape ().

Example

>>> from persista.record import Record
>>> from persista.record.analysis import compute_schema_shapes
>>> records = [
...     Record(id="a", metadata={"source": "a.pdf", "page": 1}),
...     Record(id="b", metadata={"source": "b.pdf", "page": 2}),
...     Record(id="c", metadata={"source": "c.pdf"}),
... ]
>>> compute_schema_shapes(records)
{('page', 'source'): ['a', 'b'], ('source',): ['c']}

persista.record.analysis.compute_value_frequency

compute_value_frequency(
    records: Iterable[Record],
    *,
    top_k: int = 5,
    hll_precision: int = 12,
) -> dict[str, dict[str, Any]]

Compute, for each metadata key, its most common values and an approximate cardinality (number of distinct values).

This complements compute_metadata_stats, whose unique_values_sample is a small, arbitrary (first-seen) sample rather than a ranked top-K, and does not report cardinality. Frequency counts help spot near-constant keys (a single value dominates) or skewed distributions; cardinality helps spot keys unsuitable for grouping/joins because of excessive uniqueness (e.g. UUIDs, timestamps).

Records are consumed one at a time, so this works with generators or other iterables whose full contents cannot fit in memory. Memory usage is O(number of distinct keys x number of distinct values per key) for the frequency counters (unbounded for high-cardinality keys), plus a fixed O(2 ** hll_precision) per key for the cardinality estimator.

Parameters:

Name	Type	Description	Default
records	Iterable[Record]	A list, generator, or other iterable of persista.record.Record objects. Consumed exactly once; if a generator/iterator is passed in, it will be exhausted by this call.	required
top_k	int	Number of most common values to report per key.	5
hll_precision	int	Precision passed to the internal HyperLogLog cardinality estimator for each key. Higher values give a more accurate estimate at the cost of more memory per key.	12

Returns:

Type	Description
dict[str, dict[str, Any]]	A dict mapping each metadata key to a dict with:
dict[str, dict[str, Any]]	top_values: A list of (value, count) tuples for the top_k most common values, most common first.
dict[str, dict[str, Any]]	distinct_count_estimate: The approximate number of distinct values seen for the key.

Example

>>> from persista.record import Record
>>> from persista.record.analysis import compute_value_frequency
>>> records = [
...     Record(id="a", metadata={"lang": "en"}),
...     Record(id="b", metadata={"lang": "en"}),
...     Record(id="c", metadata={"lang": "fr"}),
... ]
>>> result = compute_value_frequency(records)
>>> result["lang"]["top_values"]
[('en', 2), ('fr', 1)]

persista.record.analysis.count_approx_duplicate_records

count_approx_duplicate_records(
    records: Iterable[Record],
    *,
    expected_record_count: int = 1000000,
    fp_rate: float = 0.01,
) -> int

Count records whose metadata has probably already been seen earlier in the stream, using a Bloom filter.

Records are consumed one at a time, so this works with iterators or generators whose full contents cannot fit in memory. Unlike find_duplicate_record_ids, memory usage here is fixed up front from expected_record_count and does not grow with the number of records processed, at the cost of a tunable false-positive rate: a record may occasionally be counted as a duplicate when it is not, but a true duplicate is never missed. The hash fed to the Bloom filter is computed via :func:coola.hashing.hash_object with ignore_unhashable=True, so metadata containing values with no registered hasher is tolerated rather than raising.

Parameters:

Name	Type	Description	Default
records	Iterable[Record]	A list, generator, or other iterable of persista.record.Record objects. Consumed exactly once; if a generator/iterator is passed in, it will be exhausted by this call.	required
expected_record_count	int	Rough estimate of the total number of records (or, more precisely, unique metadata values) that will be processed. Used to size the Bloom filter for the requested fp_rate. Safe to overestimate; underestimating causes the effective false-positive rate to rise above fp_rate as more records are processed than planned for.	1000000
fp_rate	float	Target false-positive probability, once approximately expected_record_count unique metadata values have been added.	0.01

Returns:

Type	Description
int	The approximate number of records whose metadata duplicates
int	an earlier record's, as reported by the Bloom filter. Never
int	undercounts by much; may slightly overcount due to the
int	filter's false-positive rate.

Example

>>> from persista.record import Record
>>> from persista.record.analysis import count_approx_duplicate_records
>>> records = [
...     Record(id="a", metadata={"source": "a.pdf"}),
...     Record(id="b", metadata={"source": "a.pdf"}),
...     Record(id="c", metadata={"source": "b.pdf"}),
... ]
>>> count_approx_duplicate_records(records, expected_record_count=1000)
1

See Also

find_duplicate_record_ids: an exact variant using a hash map to group duplicate ids, with memory usage that grows with corpus size but produces exact (non-approximate) results.

persista.record.analysis.diff_records

diff_records(
    before: Iterable[Record], after: Iterable[Record]
) -> dict[str, list[Any]]

Diff two record snapshots (e.g. before/after a pipeline run) by id.

Useful for auditing what an ingestion or transformation step changed: which records were newly added, which disappeared, and which kept the same id but had their metadata altered.

Both before and after are consumed one at a time and fully materialized internally (as id -> metadata maps), so this works with generators or other iterables, but memory usage is O(number of records in each snapshot) - unlike the single-pass analyses elsewhere in this package, a diff inherently needs both full snapshots available for comparison.

Parameters:

Name	Type	Description	Default
before	Iterable[Record]	A list, generator, or other iterable of persista.record.Record objects representing the earlier snapshot. Consumed exactly once.	required
after	Iterable[Record]	A list, generator, or other iterable of persista.record.Record objects representing the later snapshot. Consumed exactly once.	required

Returns:

Type	Description
dict[str, list[Any]]	A dict with three keys:
dict[str, list[Any]]	added: ids present in after but not before, in the order encountered in after.
dict[str, list[Any]]	removed: ids present in before but not after, in the order encountered in before.
dict[str, list[Any]]	changed: ids present in both snapshots whose metadata differs between them, in the order encountered in after.
dict[str, list[Any]]	Ids present in both snapshots with identical metadata are
dict[str, list[Any]]	omitted entirely.

Example

>>> from persista.record import Record
>>> from persista.record.analysis import diff_records
>>> before = [
...     Record(id="a", metadata={"source": "a.pdf"}),
...     Record(id="b", metadata={"source": "b.pdf"}),
... ]
>>> after = [
...     Record(id="a", metadata={"source": "a.pdf", "page": 1}),
...     Record(id="c", metadata={"source": "c.pdf"}),
... ]
>>> diff_records(before, after)
{'added': ['c'], 'removed': ['b'], 'changed': ['a']}

persista.record.analysis.find_duplicate_record_ids

find_duplicate_record_ids(
    records: Iterable[Record],
) -> list[list[Any]]

Group record ids that share exactly the same metadata.

Records are consumed one at a time, so this works with generators or other iterables whose full contents cannot fit in memory. Only a hash of each record's metadata is retained (rather than the full metadata itself), making this more memory-efficient for large corpora. The hash is computed via :func:coola.hashing.hash_object with ignore_unhashable=True, so metadata containing values with no registered hasher is tolerated (grouped by a deterministic placeholder) rather than raising.

Parameters:

Name	Type	Description	Default
records	Iterable[Record]	A list, generator, or other iterable of persista.record.Record objects. Consumed exactly once; if a generator/iterator is passed in, it will be exhausted by this call.	required

Returns:

Type	Description
list[list[Any]]	A list of groups of record ids whose metadata is
list[list[Any]]	identical. Each group contains two or more ids, in the order
list[list[Any]]	their records were encountered; groups are in the order their
list[list[Any]]	first member was encountered. Records with unique metadata are
list[list[Any]]	not included in the result.

Example

>>> from persista.record import Record
>>> from persista.record.analysis import find_duplicate_record_ids
>>> records = [
...     Record(id="a", metadata={"source": "a.pdf"}),
...     Record(id="b", metadata={"source": "a.pdf"}),
...     Record(id="c", metadata={"source": "b.pdf"}),
... ]
>>> find_duplicate_record_ids(records)
[['a', 'b']]

See Also

count_approx_duplicate_records: an approximate variant using a Bloom filter, with fixed (O(1)) memory usage, better suited to corpora too large for the exact hash groups above to fit in memory.

persista.record.analysis.find_id_collisions

find_id_collisions(
    records: Iterable[Record],
) -> dict[Any, list[dict[str, Any]]]

Find record ids that are reused across records with different metadata.

find_duplicate_record_ids groups different ids that share identical metadata. This is the complementary check: it flags the same id appearing more than once with different metadata, which usually indicates an id-generation bug (e.g. a non-unique or colliding hash) rather than a legitimate duplicate.

An id that appears multiple times with exactly the same metadata each time is not considered a collision - it is treated as the same record encountered more than once.

Records are consumed one at a time, so this works with generators or other iterables whose full contents cannot fit in memory. Memory usage is O(number of distinct ids x number of distinct metadata variants per id).

Parameters:

Name	Type	Description	Default
records	Iterable[Record]	A list, generator, or other iterable of persista.record.Record objects. Consumed exactly once; if a generator/iterator is passed in, it will be exhausted by this call.	required

Returns:

Type	Description
dict[Any, list[dict[str, Any]]]	A dict mapping each colliding record id to the list of
dict[Any, list[dict[str, Any]]]	distinct metadata values observed for it, in the order
dict[Any, list[dict[str, Any]]]	they were first encountered. Ids observed with only one
dict[Any, list[dict[str, Any]]]	distinct metadata value (whether seen once or many times) are
dict[Any, list[dict[str, Any]]]	not included.

Example

>>> from persista.record import Record
>>> from persista.record.analysis import find_id_collisions
>>> records = [
...     Record(id="a", metadata={"source": "x.pdf"}),
...     Record(id="a", metadata={"source": "y.pdf"}),
...     Record(id="b", metadata={"source": "z.pdf"}),
...     Record(id="b", metadata={"source": "z.pdf"}),
... ]
>>> find_id_collisions(records)
{'a': [{'source': 'x.pdf'}, {'source': 'y.pdf'}]}

persista.record.analysis.find_near_duplicate_record_ids

find_near_duplicate_record_ids(
    records: Iterable[Record],
    *,
    num_hashes: int = 32,
    num_bands: int = 8,
) -> list[list[Any]]

Group record ids whose metadata is approximately similar, using MinHash signatures and locality-sensitive hashing (LSH) banding.

Unlike find_duplicate_record_ids (which only groups records with byte-for-byte identical metadata), this catches records whose metadata overlaps substantially but not completely - e.g. the same document re-ingested with one extra or corrected field, or metadata differing only in a timestamp key.

Each record's metadata is treated as a set of "key=value" tokens. A num_hashes-sized MinHash signature approximates the Jaccard similarity between two records' token sets; signatures are then split into num_bands bands, and any two records sharing an identical band are considered candidate near-duplicates and merged into the same group (via union-find). More bands (with fewer rows each) increases recall (catches less similar pairs) at the cost of more false positives; fewer bands does the opposite.

Records are consumed one at a time, so this works with generators or other iterables whose full contents cannot fit in memory for the signature computation itself, though - like find_duplicate_record_ids - the signatures and grouping structure are retained for the whole corpus, so memory usage grows with the number of records (O(number of records x num_hashes)).

Parameters:

Name	Type	Description	Default
records	Iterable[Record]	A list, generator, or other iterable of persista.record.Record objects. Consumed exactly once; if a generator/iterator is passed in, it will be exhausted by this call.	required
num_hashes	int	Number of hash functions in each MinHash signature. Must be evenly divisible by num_bands.	32
num_bands	int	Number of LSH bands the signature is split into. Must evenly divide num_hashes.	8

Returns:

Type	Description
list[list[Any]]	A list of groups of record ids considered near-duplicates
list[list[Any]]	of each other, in the order their records were encountered.
list[list[Any]]	Each group contains two or more ids. Records with no
list[list[Any]]	near-duplicate are not included. Records with empty metadata
list[list[Any]]	are never grouped with each other or with anything else.

Raises:

Type	Description
ValueError	If num_hashes is not evenly divisible by num_bands.

Example

>>> from persista.record import Record
>>> from persista.record.analysis import find_near_duplicate_record_ids
>>> records = [
...     Record(id="a", metadata={"source": "a.pdf", "page": 1, "lang": "en"}),
...     Record(id="b", metadata={"source": "a.pdf", "page": 1, "lang": "fr"}),
...     Record(id="c", metadata={"source": "z.pdf", "page": 9}),
... ]
>>> find_near_duplicate_record_ids(records, num_hashes=4, num_bands=4)
[['a', 'b']]

See Also

find_duplicate_record_ids: an exact variant that only groups records with identical metadata.

persista.record.analysis.find_orphan_references

find_orphan_references(
    records: Iterable[Record], reference_key: str
) -> list[Any]

Find records whose metadata references another record id that does not exist in the corpus.

Some corpora encode relationships between records via a metadata field holding another record's id (e.g. a "parent_id" for a chunk pointing back at the document it was split from). This detects dangling references: values of reference_key that do not match any id present in records.

This requires two passes over records (once to collect all ids, once to check references). records is materialized into a list up front if it is not already one, so a single-use generator or iterator works too, at the cost of holding all records in memory at once.

Parameters:

Name	Type	Description	Default
records	Iterable[Record]	A list, tuple, generator, or other iterable of persista.record.Record objects.	required
reference_key	str	The metadata key holding the referenced record id. Records missing this key, or whose value is None, are not considered orphans (there is nothing to check).	required

Returns:

Type	Description
list[Any]	A list of record ids whose reference_key metadata
list[Any]	value does not match any record id in the corpus, in the
list[Any]	order their records were encountered.

Example

>>> from persista.record import Record
>>> from persista.record.analysis import find_orphan_references
>>> records = [
...     Record(id="doc1", metadata={}),
...     Record(id="chunk1", metadata={"parent_id": "doc1"}),
...     Record(id="chunk2", metadata={"parent_id": "missing_doc"}),
...     Record(id="chunk3", metadata={}),
... ]
>>> find_orphan_references(records, reference_key="parent_id")
['chunk2']

persista.record.analysis.print_metadata_stats_report

print_metadata_stats_report(
    stats: dict[str, Any],
    *,
    title: str = "Metadata Report",
    console: Console | None = None,
) -> None

Render a record-metadata-stats report (from compute_metadata_stats) as an overview line, a summary table, and a per-key breakdown table, inside a cyan panel that shrinks to fit its content rather than stretching to the full terminal width.

This function orchestrates the report layout by delegating each section to a dedicated builder: _build_overview_line (record/key count summary), _build_summary_table (keys-per-record / data quality), and _build_per_key_table (one row per metadata key, with sampled values and data-quality flags).

Parameters:

Name	Type	Description	Default
stats	dict[str, Any]	The dict returned by compute_metadata_stats (or MetadataStats.to_dict). Expected to contain a count key; all other keys are read defensively via [...]/ .get(...), so a partial dict for an empty corpus (e.g. count=0) still renders correctly.	required
title	str	Panel title shown above the report, at the top-left of the panel border.	'Metadata Report'
console	Console | None	An existing rich.console.Console to print to. Defaults to the shared rich console via get_console().	None

Returns:

Type	Description
None	None. Output is printed directly to console.

Examples:

>>> stats = {
...     "count": 2, "missing_metadata_count": 0,
...     "avg_keys": 1.5, "min_keys": 1, "max_keys": 2,
...     "distinct_keys_seen": 2,
...     "per_key": {
...         "source": {
...             "present_in_records": 2, "missing_in_records": 0,
...             "value_types": ["str"], "none_or_empty_count": 0,
...             "unique_values_sample": ["a.pdf", "b.pdf"],
...             "unique_values_sample_truncated": False,
...         },
...     },
... }
>>> print_metadata_stats_report(stats)

persista.record.store

Contain record store utilities.

persista.record.store.BaseRecordStore

Bases: ABC

Abstract base class for record stores.

Defines the common interface that all record store implementations must provide. The API mirrors :class:persista.store.BaseStore, with :class:~persista.record.Record instances (keyed by their id) taking the place of key-value pairs.

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:open on entry and :meth:close on exit.

Constructing a record store does not connect to the underlying backend: implementations must defer that to :meth:open/ :meth:aopen, so every other method (including :meth:close) raises until the store has been opened, either explicitly or via the context manager.

persista.record.store.BaseRecordStore.closed abstractmethod property

closed: bool

Indicate whether the store is closed.

Returns:

Type	Description
bool	True if the store has been closed, False if it is
bool	open and ready to use.

persista.record.store.BaseRecordStore.aclear abstractmethod async

aclear() -> None

Async equivalent of :meth:clear.

persista.record.store.BaseRecordStore.aclose abstractmethod async

aclose() -> None

Async equivalent of :meth:close.

persista.record.store.BaseRecordStore.acontains abstractmethod async

acontains(record_id: str) -> bool

Async equivalent of :meth:contains.

persista.record.store.BaseRecordStore.acontains_many abstractmethod async

acontains_many(record_ids: list[str]) -> list[bool]

Async equivalent of :meth:contains_many.

persista.record.store.BaseRecordStore.acount abstractmethod async

acount() -> int

Async equivalent of :meth:count.

persista.record.store.BaseRecordStore.adelete abstractmethod async

adelete(record_id: str) -> None

Async equivalent of :meth:delete.

persista.record.store.BaseRecordStore.adelete_many abstractmethod async

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

Async equivalent of :meth:delete_many.

persista.record.store.BaseRecordStore.afilter abstractmethod async

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

Async equivalent of :meth:filter.

persista.record.store.BaseRecordStore.aget abstractmethod async

aget(record_id: str) -> Record | None

Async equivalent of :meth:get.

persista.record.store.BaseRecordStore.aget_many abstractmethod async

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

Async equivalent of :meth:get_many.

persista.record.store.BaseRecordStore.aiter_batches abstractmethod

aiter_batches(
    batch_size: int = 32,
) -> AsyncIterator[list[Record]]

Async equivalent of :meth:iter_batches.

persista.record.store.BaseRecordStore.akeys abstractmethod

akeys() -> AsyncIterator[str]

Async equivalent of :meth:keys.

persista.record.store.BaseRecordStore.aopen abstractmethod async

aopen() -> None

Async equivalent of :meth:open.

persista.record.store.BaseRecordStore.aset_many abstractmethod async

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

Async equivalent of :meth:set_many.

persista.record.store.BaseRecordStore.avalues async

avalues(batch_size: int = 32) -> AsyncIterator[Record]

Async equivalent of :meth:values.

persista.record.store.BaseRecordStore.clear abstractmethod

clear() -> None

Remove every record from the store.

This is equivalent to resetting the store to empty, without closing it.

persista.record.store.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.

persista.record.store.BaseRecordStore.contains abstractmethod

contains(record_id: str) -> bool

Check if a record ID exists in the store.

Parameters:

Name	Type	Description	Default
record_id	str	The record ID to check.	required

Returns:

Type	Description
bool	True if the record exists in the store, False
bool	otherwise.

persista.record.store.BaseRecordStore.contains_many abstractmethod

contains_many(record_ids: list[str]) -> list[bool]

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
list[bool]	A list of booleans, in the same order as record_ids,
list[bool]	where each entry is True if the corresponding record
list[bool]	ID exists in the store and False otherwise.

persista.record.store.BaseRecordStore.count abstractmethod

count() -> int

Return the total number of records in the store.

Returns:

Type	Description
int	The number of records currently stored.

persista.record.store.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

persista.record.store.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

persista.record.store.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:~persista.record.Record
list[Record]	instances.

persista.record.store.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:~persista.record.Record, or None if not
	Record | None	found.

persista.record.store.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:

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

persista.record.store.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:values: 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
list[Record]	elements, in the same order as :meth:values. The last
list[Record]	batch may contain fewer than batch_size records.

persista.record.store.BaseRecordStore.keys abstractmethod

keys() -> Iterator[str]

Iterate over the IDs of all records in the store.

Yields:

Type	Description
str	Every record ID currently in the store.

persista.record.store.BaseRecordStore.open abstractmethod

open() -> None

Connect to the underlying backend and prepare the store for use (e.g. open a database connection, create a directory).

The constructor must not do this itself: implementations connect lazily, only once open() (or :meth:__enter__) is called. Implementations should make repeated calls to open() safe (i.e. idempotent), since a store may be reopened after :meth:close.

persista.record.store.BaseRecordStore.set_many abstractmethod

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

Add or replace records in the store.

Records whose id already exists should be replaced (upsert semantics).

Parameters:

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

persista.record.store.BaseRecordStore.values

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

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

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:~persista.record.Record at a time, in the
	Record	same order as :meth:iter_batches.

persista.record.store.DuckDBRecordStore

Bases: _UntypedRecordStore

Implement a :class:~persista.record.store.RecordStore backed by a DuckDB database, with record metadata stored as JSON.

This is a convenience wrapper around :class:persista.store.TypedDuckDBStore with no fixed schema, so every metadata field is stored in the JSON overflow column. Use :class:TypedDuckDBRecordStore instead when metadata fields should be stored as their own SQL columns.

Parameters:

Name	Type	Description	Default
database	Path | str	The path to the DuckDB database file, or ":memory:" for an in-memory database.	':memory:'
**kwargs	Any	Additional keyword arguments passed to :class:persista.store.TypedDuckDBStore.	{}

Example

>>> from persista.record import Record
>>> from persista.record.store import DuckDBRecordStore
>>> with DuckDBRecordStore() as store:
...     store.set_many([Record(id="1", metadata={"author": "Alice"})])
...     store.get("1")
...
Record(id='1', metadata={'author': 'Alice'})

persista.record.store.InMemoryRecordStore

Bases: RecordStore

Implement a :class:~persista.record.store.RecordStore backed by a plain in-memory dictionary store.

This is a convenience wrapper around :class:persista.store.InMemoryStore, useful for tests, examples, and other scenarios where records do not need to be persisted across processes.

Example

>>> from persista.record import Record
>>> from persista.record.store import InMemoryRecordStore
>>> with InMemoryRecordStore() as store:
...     store.set_many([Record(id="1", metadata={"author": "Alice"})])
...     store.get("1")
...
Record(id='1', metadata={'author': 'Alice'})

persista.record.store.RecordStore

Bases: BaseRecordStore, MultilineDisplayMixin

Implement a record store backed by a :class:persista.store.BaseStore.

Records are keyed by their id in the underlying key-value store, and a record's metadata dict is stored directly as the value, so the id itself is not duplicated in the stored value.

Parameters:

Name	Type	Description	Default
store	BaseStore	The underlying key-value store.	required

Example

>>> from persista.record import Record
>>> from persista.record.store import RecordStore
>>> from persista.store import InMemoryStore
>>> with RecordStore(InMemoryStore()) as store:
...     store.set_many([Record(id="1", metadata={"author": "Alice"})])
...     store.get("1")
...
Record(id='1', metadata={'author': 'Alice'})

persista.record.store.RecordStore.afilter async

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

Async equivalent of :meth:filter.

Note

Same fast-path/full-scan behavior as :meth:filter; see its Note.

persista.record.store.RecordStore.filter

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

See :meth:BaseRecordStore.filter.

Note

When the underlying store exposes filter_items (e.g. Typed*Store variants whose metadata fields are stored as their own SQL columns), metadata_filters is pushed down to it, letting the store answer via a SQL WHERE clause instead of a full scan. Otherwise this falls back to a full scan of the underlying store (via :meth:~persista.store.BaseStore.iter_batches), filtering records in Python.

persista.record.store.SQLiteRecordStore

Bases: _UntypedRecordStore

Implement a :class:~persista.record.store.RecordStore backed by a SQLite database, with record metadata stored as JSON.

This is a convenience wrapper around :class:persista.store.TypedSQLiteStore with no fixed schema, so every metadata field is stored in the JSON overflow column. Use :class:TypedSQLiteRecordStore instead when metadata fields should be stored as their own SQL columns.

Parameters:

Name	Type	Description	Default
database	Path | str	The path to the SQLite database file, or ":memory:" for an in-memory database.	':memory:'
**kwargs	Any	Additional keyword arguments passed to :class:persista.store.TypedSQLiteStore.	{}

Example

>>> from persista.record import Record
>>> from persista.record.store import SQLiteRecordStore
>>> with SQLiteRecordStore() as store:
...     store.set_many([Record(id="1", metadata={"author": "Alice"})])
...     store.get("1")
...
Record(id='1', metadata={'author': 'Alice'})

persista.record.store.TypedDuckDBRecordStore

Bases: _TypedRecordStore

Implement a :class:~persista.record.store.RecordStore backed by a DuckDB database, with each record metadata field stored as its own typed SQL column.

This is a convenience wrapper around :class:persista.store.TypedDuckDBStore configured with one column per metadata field declared in metadata_schema.

Use this store when record metadata follows a known, fixed schema and individual metadata fields should be queryable as SQL columns. Use :class:DuckDBRecordStore instead when records may have arbitrary or varying metadata.

Parameters:

Name	Type	Description	Default
database	Path | str	The path to the DuckDB database file, or ":memory:" for an in-memory database.	':memory:'
metadata_schema	dict[str, str] | None	A mapping from metadata field name to its SQL column type declaration (e.g. {"author": "TEXT"}). None is equivalent to an empty mapping, i.e. no metadata columns beyond the JSON overflow column.	None
**kwargs	Any	Additional keyword arguments passed to :class:persista.store.TypedDuckDBStore.	{}

Example

>>> from persista.record import Record
>>> from persista.record.store import TypedDuckDBRecordStore
>>> with TypedDuckDBRecordStore(metadata_schema={"author": "TEXT"}) as store:
...     store.set_many([Record(id="1", metadata={"author": "Alice"})])
...     store.get("1")
...
Record(id='1', metadata={'author': 'Alice'})

persista.record.store.TypedSQLiteRecordStore

Bases: _TypedRecordStore

Implement a :class:~persista.record.store.RecordStore backed by a SQLite database, with each record metadata field stored as its own typed SQL column.

This is a convenience wrapper around :class:persista.store.TypedSQLiteStore configured with one column per metadata field declared in metadata_schema.

Use this store when record metadata follows a known, fixed schema and individual metadata fields should be queryable as SQL columns. Use :class:SQLiteRecordStore instead when records may have arbitrary or varying metadata.

Parameters:

Name	Type	Description	Default
database	Path | str	The path to the SQLite database file, or ":memory:" for an in-memory database.	':memory:'
metadata_schema	dict[str, str] | None	A mapping from metadata field name to its SQL column type declaration (e.g. {"author": "TEXT"}). None is equivalent to an empty mapping, i.e. no metadata columns beyond the JSON overflow column.	None
**kwargs	Any	Additional keyword arguments passed to :class:persista.store.TypedSQLiteStore.	{}

Example

>>> from persista.record import Record
>>> from persista.record.store import TypedSQLiteRecordStore
>>> with TypedSQLiteRecordStore(metadata_schema={"author": "TEXT"}) as store:
...     store.set_many([Record(id="1", metadata={"author": "Alice"})])
...     store.get("1")
...
Record(id='1', metadata={'author': 'Alice'})

persista.record.store.check_is_closed

check_is_closed(store: BaseRecordStore) -> None

Check that a record store is closed, raising if it is not.

Parameters:

Name	Type	Description	Default
store	BaseRecordStore	The :class:~persista.record.store.base.BaseRecordStore to check.	required

Raises:

Type	Description
RuntimeError	If store is open.

Example

>>> from persista.record.store import InMemoryRecordStore
>>> from persista.record.store.validation import check_is_closed
>>> store = InMemoryRecordStore()
>>> check_is_closed(store)

persista.record.store.check_is_open

check_is_open(store: BaseRecordStore) -> None

Check that a record store is open, raising if it is not.

Parameters:

Name	Type	Description	Default
store	BaseRecordStore	The :class:~persista.record.store.base.BaseRecordStore to check.	required

Raises:

Type	Description
RuntimeError	If store is closed.

Example

>>> from persista.record.store import InMemoryRecordStore
>>> from persista.record.store.validation import check_is_open
>>> store = InMemoryRecordStore()
>>> with store:
...     check_is_open(store)
...

persista.record.store.resolve_record_store

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

Resolve a :class:~persista.record.store.BaseRecordStore instance from an existing object or a configuration dictionary.

If store is already a :class:~persista.record.store.BaseRecordStore instance it is returned as-is. If it is a :class:dict, it is treated as an objectory factory configuration and instantiated via :func:objectory.factory. See :func:~coola.factory.resolve_object for details.

Parameters:

Name	Type	Description	Default
store	BaseRecordStore | dict[str, Any]	Either a fully configured :class:~persista.record.store.BaseRecordStore instance, or a :class:dict containing an objectory factory specification (must include a "_target_" key pointing to the fully-qualified class name).	required

Returns:

Type	Description
BaseRecordStore	A configured :class:~persista.record.store.BaseRecordStore
BaseRecordStore	instance.

Raises:

Type	Description
TypeError	If the resolved object is not a :class:~persista.record.store.BaseRecordStore instance.

Example

>>> from persista.record.store import InMemoryRecordStore, resolve_record_store
>>> # From an existing instance:
>>> store = resolve_record_store(InMemoryRecordStore())
>>> # From a configuration dictionary:
>>> store = resolve_record_store({"_target_": "persista.record.store.InMemoryRecordStore"})

persista.record.store.factory

Contain factories for record stores.

persista.record.store.factory.BaseRecordStoreFactory

Bases: ABC

Abstract base class for :class:~persista.record.store.BaseRecordStore factories.

Subclasses implement :meth:make_record_store to instantiate and return a configured :class:~persista.record.store.BaseRecordStore object. This pattern decouples record store creation from the rest of the codebase, making it easy to swap how a record store is built (e.g. a shared instance vs. a fresh one per call) without changing call sites.

Example

>>> from persista.record.store import BaseRecordStore, InMemoryRecordStore
>>> from persista.record.store.factory import BaseRecordStoreFactory
>>> class MyRecordStoreFactory(BaseRecordStoreFactory):
...     def make_record_store(self) -> BaseRecordStore:
...         return InMemoryRecordStore()
...
>>> factory = MyRecordStoreFactory()
>>> store = factory.make_record_store()

persista.record.store.factory.BaseRecordStoreFactory.make_record_store abstractmethod

make_record_store() -> BaseRecordStore

Create and return a configured BaseRecordStore instance.

Returns:

Name	Type	Description
A	BaseRecordStore	class:~persista.record.store.BaseRecordStore
	BaseRecordStore	instance ready for use.

persista.record.store.factory.ConfigurableRecordStoreFactory

Bases: BaseRecordStoreFactory, MultilineDisplayMixin

A concrete BaseRecordStore factory that accepts either a pre-built :class:~persista.record.store.BaseRecordStore instance or a configuration dictionary.

When a dict is provided it is resolved at each :meth:make_record_store call via :func:~persista.record.store.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:~persista.record.store.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 persista.record.store import InMemoryRecordStore
>>> from persista.record.store.factory import ConfigurableRecordStoreFactory
>>> factory = ConfigurableRecordStoreFactory(InMemoryRecordStore())
>>> store = factory.make_record_store()

persista.record.store.factory.DuckDBRecordStoreFactory

Bases: _UntypedRecordStoreFactory

Implement a record store factory that builds a new :class:~persista.record.store.DuckDBRecordStore on each call.

Parameters:

Name	Type	Description	Default
database	Path | str	The path to the DuckDB database file, or ":memory:" for an in-memory database.	':memory:'
**kwargs	Any	Additional keyword arguments passed to :class:~persista.record.store.DuckDBRecordStore.	{}

Example

>>> from persista.record.store.factory import DuckDBRecordStoreFactory
>>> factory = DuckDBRecordStoreFactory()
>>> store = factory.make_record_store()

persista.record.store.factory.InMemoryRecordStoreFactory

Bases: BaseRecordStoreFactory, MultilineDisplayMixin

Implement a record store factory that builds a new :class:~persista.record.store.InMemoryRecordStore on each call.

Example

>>> from persista.record.store.factory import InMemoryRecordStoreFactory
>>> factory = InMemoryRecordStoreFactory()
>>> store = factory.make_record_store()

persista.record.store.factory.RecordStoreFactory

Bases: BaseRecordStoreFactory, MultilineDisplayMixin

A concrete BaseRecordStore factory that wraps a pre-built :class:~persista.record.store.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:~persista.record.store.BaseRecordStore instance to return from :meth:make_record_store.	required

Example

>>> from persista.record.store import InMemoryRecordStore
>>> from persista.record.store.factory import RecordStoreFactory
>>> factory = RecordStoreFactory(InMemoryRecordStore())
>>> store = factory.make_record_store()

persista.record.store.factory.SQLiteRecordStoreFactory

Bases: _UntypedRecordStoreFactory

Implement a record store factory that builds a new :class:~persista.record.store.SQLiteRecordStore on each call.

Parameters:

Name	Type	Description	Default
database	Path | str	The path to the SQLite database file, or ":memory:" for an in-memory database.	':memory:'
**kwargs	Any	Additional keyword arguments passed to :class:~persista.record.store.SQLiteRecordStore.	{}

Example

>>> from persista.record.store.factory import SQLiteRecordStoreFactory
>>> factory = SQLiteRecordStoreFactory()
>>> store = factory.make_record_store()

persista.record.store.factory.StoreRecordStoreFactory

Bases: BaseRecordStoreFactory, MultilineDisplayMixin

A concrete record store factory that builds its backing store from a :class:~persista.store.factory.BaseStoreFactory.

Use this when the underlying key-value store needs to be freshly created (e.g. a new connection, a new in-memory dict) each time a :class:~persista.record.store.RecordStore is requested, rather than sharing one store instance across every record store.

Parameters:

Name	Type	Description	Default
store_factory	BaseStoreFactory	The factory used to create the backing key-value store passed to each :class:~persista.record.store.RecordStore built by :meth:make_record_store.	required

Example

>>> from persista.record.store.factory import StoreRecordStoreFactory
>>> from persista.store.factory import InMemoryStoreFactory
>>> factory = StoreRecordStoreFactory(InMemoryStoreFactory())
>>> store = factory.make_record_store()
>>> store.open()

persista.record.store.factory.TypedDuckDBRecordStoreFactory

Bases: _TypedRecordStoreFactory

Implement a record store factory that builds a new :class:~persista.record.store.TypedDuckDBRecordStore on each call.

Parameters:

Name	Type	Description	Default
database	Path | str	The path to the DuckDB database file, or ":memory:" for an in-memory database.	':memory:'
metadata_schema	dict[str, str] | None	A mapping from metadata field name to its SQL column type declaration (e.g. {"author": "TEXT"}). None is equivalent to an empty mapping, i.e. no metadata columns beyond the JSON overflow column.	None
**kwargs	Any	Additional keyword arguments passed to :class:~persista.record.store.TypedDuckDBRecordStore.	{}

Example

>>> from persista.record.store.factory import TypedDuckDBRecordStoreFactory
>>> factory = TypedDuckDBRecordStoreFactory(metadata_schema={"author": "TEXT"})
>>> store = factory.make_record_store()

persista.record.store.factory.TypedSQLiteRecordStoreFactory

Bases: _TypedRecordStoreFactory

Implement a record store factory that builds a new :class:~persista.record.store.TypedSQLiteRecordStore on each call.

Parameters:

Name	Type	Description	Default
database	Path | str	The path to the SQLite database file, or ":memory:" for an in-memory database.	':memory:'
metadata_schema	dict[str, str] | None	A mapping from metadata field name to its SQL column type declaration (e.g. {"author": "TEXT"}). None is equivalent to an empty mapping, i.e. no metadata columns beyond the JSON overflow column.	None
**kwargs	Any	Additional keyword arguments passed to :class:~persista.record.store.TypedSQLiteRecordStore.	{}

Example

>>> from persista.record.store.factory import TypedSQLiteRecordStoreFactory
>>> factory = TypedSQLiteRecordStoreFactory(metadata_schema={"author": "TEXT"})
>>> store = factory.make_record_store()