dp3.database.snapshots

TypedSnapshotCollection

TypedSnapshotCollection(db: Database, entity_type: str, db_config: MongoConfig, model_spec: ModelSpec, snapshots_config: dict)

Bases: ABC

Snapshot collection handler with eid type awareness.

Source code in dp3/database/snapshots.py

def __init__(
    self,
    db: Database,
    entity_type: str,
    db_config: MongoConfig,
    model_spec: ModelSpec,
    snapshots_config: dict,
):
    self._db = db.with_options(codec_options=get_codec_options())

    if entity_type not in model_spec.entities:
        raise ValueError(f"Entity type '{entity_type}' not found in model spec")
    self.entity_type = entity_type
    self._col_name = f"{entity_type}#snapshots"
    self._os_col_name = f"{entity_type}#snapshots_oversized"

    self.attr_specs: dict[str, AttrSpecType] = model_spec.entity_attributes[entity_type]

    self.log = logging.getLogger(f"EntityDatabase.SnapshotCollection[{entity_type}]")

    self._normal_snapshot_eids = set()
    self._oversized_snapshot_eids = set()
    self._snapshot_bucket_size = db_config.storage.snapshot_bucket_size
    self._bucket_delta = self._get_snapshot_bucket_delta(snapshots_config)
    self._server_max_bson_size: int | None = None

get_latest_one

get_latest_one(eid: AnyEidT) -> dict

Get latest snapshot of given eid.

If doesn't exist, returns {}.

Source code in dp3/database/snapshots.py

def get_latest_one(self, eid: AnyEidT) -> dict:
    """Get latest snapshot of given eid.

    If doesn't exist, returns {}.
    """
    return (
        self._col().find_one(self._filter_from_eid(eid), {"last": 1}, sort=[("_id", -1)]) or {}
    ).get("last", {})

find_latest

find_latest(fulltext_filters: dict[str, str] | None = None, generic_filter: dict[str, Any] | None = None) -> Cursor

Find latest snapshots of given etype.

This method is useful for displaying data on web.

Returns only documents matching generic_filter and fulltext_filters (dictionary attribute - fulltext filter). Fulltext filters are interpreted as regular expressions. Only string values may be filtered this way. There's no validation that queried attribute can be fulltext filtered. Only plain and observation attributes with string-based data types can be queried. Array and set data types are supported as well as long as they are not multi value at the same time. If you need to filter EIDs, ensure the EID is string, then use attribute eid. Otherwise, use generic filter.

Generic filter allows filtering using generic MongoDB query (including $and, $or, $lt, etc.). For querying non-JSON-native types, you can use magic strings, such as "$$IPv4{<ip address>}" for IPv4 addresses. The full spec with examples is in the magic strings module.

Generic and fulltext filters are merged - fulltext overrides conflicting keys.

May raise SnapshotCollectionError if query is invalid.

Returns only documents matching generic_filter and fulltext_filters, does not count them.

Source code in dp3/database/snapshots.py

def find_latest(
    self,
    fulltext_filters: dict[str, str] | None = None,
    generic_filter: dict[str, Any] | None = None,
) -> Cursor:
    """Find latest snapshots of given `etype`.

    This method is useful for displaying data on web.

    Returns only documents matching `generic_filter` and `fulltext_filters`
    (dictionary attribute - fulltext filter).
    Fulltext filters are interpreted as regular expressions.
    Only string values may be filtered this way. There's no validation that queried attribute
    can be fulltext filtered.
    Only plain and observation attributes with string-based data types can be queried.
    Array and set data types are supported as well as long as they are not multi value
    at the same time.
    If you need to filter EIDs, ensure the EID is string, then use attribute `eid`.
    Otherwise, use generic filter.

    Generic filter allows filtering using generic MongoDB query (including `$and`, `$or`,
    `$lt`, etc.).
    For querying non-JSON-native types, you can use magic strings, such as
    `"$$IPv4{<ip address>}"` for IPv4 addresses. The full spec with examples is in the
    [magic strings module][dp3.database.magic].

    Generic and fulltext filters are merged - fulltext overrides conflicting keys.

    May raise `SnapshotCollectionError` if query is invalid.

    Returns only documents matching `generic_filter` and `fulltext_filters`,
    does not count them.
    """
    query = self._prepare_latest_query(fulltext_filters or {}, generic_filter or {})
    try:
        return self._col().find(query, {"last": 1}).sort([("_id", pymongo.ASCENDING)])
    except OperationFailure as e:
        raise SnapshotCollectionError(f"Query is invalid: {e}") from e

count_latest

count_latest(fulltext_filters: dict[str, str] | None = None, generic_filter: dict[str, Any] | None = None) -> int

Count latest snapshots of given etype.

See find_latest for more information.

Returns only count of documents matching generic_filter and fulltext_filters.

Note that this method may take much longer than find_latest on larger databases, as it does count all documents, not just return the first few.

Source code in dp3/database/snapshots.py

def count_latest(
    self,
    fulltext_filters: dict[str, str] | None = None,
    generic_filter: dict[str, Any] | None = None,
) -> int:
    """Count latest snapshots of given `etype`.

    See [`find_latest`][dp3.database.snapshots.SnapshotCollectionContainer.find_latest]
    for more information.

    Returns only count of documents matching `generic_filter` and `fulltext_filters`.

    Note that this method may take much longer than `find_latest` on larger databases,
    as it does count all documents, not just return the first few.
    """
    query = self._prepare_latest_query(fulltext_filters or {}, generic_filter or {})
    try:
        return self._col().count_documents(query)
    except OperationFailure as e:
        raise SnapshotCollectionError(f"Query is invalid: {e}") from e

get_by_eid

get_by_eid(eid: AnyEidT, t1: datetime | None = None, t2: datetime | None = None, skip: int = 0, limit: int = 0) -> Cursor | CommandCursor

Get all (or filtered) snapshots of given eid.

This method is useful for displaying eid's history on web.

Parameters:

Name	Type	Description	Default
eid	AnyEidT	id of entity, to which data-points correspond	required
t1	datetime | None	left value of time interval (inclusive)	None
t2	datetime | None	right value of time interval (inclusive)	None
skip	int	number of snapshots to skip	0
limit	int	max number of snapshots to return, 0 means no limit	0

Source code in dp3/database/snapshots.py

def get_by_eid(
    self,
    eid: AnyEidT,
    t1: datetime | None = None,
    t2: datetime | None = None,
    skip: int = 0,
    limit: int = 0,
) -> Cursor | CommandCursor:
    """Get all (or filtered) snapshots of given `eid`.

    This method is useful for displaying `eid`'s history on web.

    Args:
        eid: id of entity, to which data-points correspond
        t1: left value of time interval (inclusive)
        t2: right value of time interval (inclusive)
        skip: number of snapshots to skip
        limit: max number of snapshots to return, `0` means no limit
    """
    snapshot_col = self._col()

    # Find out if the snapshot is oversized
    doc = (
        snapshot_col.find(self._filter_from_eid(eid), {"oversized": 1})
        .sort([("_id", -1)])
        .limit(1)
    )
    doc = next(doc, None)
    if doc and doc.get("oversized", False):
        return self._get_oversized(eid, t1, t2, skip, limit)

    query = {"_time_created": {}}
    pipeline = [
        {"$match": self._filter_from_eid(eid)},
        {"$unwind": "$history"},
        {"$replaceRoot": {"newRoot": "$history"}},
    ]

    # Filter by date
    if t1:
        query["_time_created"]["$gte"] = t1
    if t2:
        query["_time_created"]["$lte"] = t2

    # Unset if empty
    if query["_time_created"]:
        pipeline.append({"$match": query})

    pipeline.append({"$sort": {"_time_created": pymongo.ASCENDING}})
    if skip:
        pipeline.append({"$skip": skip})
    if limit:
        pipeline.append({"$limit": limit})
    return snapshot_col.aggregate(pipeline)

get_distinct_val_count

get_distinct_val_count(attr: str) -> dict[Any, int]

Counts occurrences of distinct values of given attribute in snapshots.

Returns dictionary mapping value -> count.

Works for all plain and observation data types except dict and json.

Source code in dp3/database/snapshots.py

def get_distinct_val_count(self, attr: str) -> dict[Any, int]:
    """Counts occurrences of distinct values of given attribute in snapshots.

    Returns dictionary mapping value -> count.

    Works for all plain and observation data types except `dict` and `json`.
    """
    # Get attribute specification
    try:
        attr_spec = self.attr_specs[attr]
    except KeyError as e:
        raise SnapshotCollectionError(f"Attribute '{attr}' does not exist") from e

    if attr_spec.t not in AttrType.PLAIN | AttrType.OBSERVATIONS:
        raise SnapshotCollectionError(f"Attribute '{attr}' isn't plain or observations")

    # Attribute data type must be primitive, array<T> or set<T>
    if any(needle in attr_spec.data_type.root for needle in ("dict", "json")):
        raise SnapshotCollectionError(
            f"Data type '{attr_spec.data_type}' of attribute '{attr}' is not processable"
        )

    # Build aggregation query
    attr_path = "$last." + attr
    unwinding = []

    # Unwind array-like and multi value attributes
    # If attribute is multi value array, unwind twice
    if "array" in attr_spec.data_type.root or "set" in attr_spec.data_type.root:
        unwinding.append({"$unwind": attr_path})
    if attr_spec.t == AttrType.OBSERVATIONS and attr_spec.multi_value:
        unwinding.append({"$unwind": attr_path})

    # Group
    agg_query_group_id = attr_path
    if "link" in attr_spec.data_type.root:
        agg_query_group_id += ".eid"

    agg_query = [
        {"$match": {"latest": True}},
        *unwinding,
        {"$group": {"_id": agg_query_group_id, "count": {"$sum": 1}}},
        {"$sort": {"_id": 1, "count": -1}},
    ]
    # Run aggregation
    distinct_counts_cur = self._col().aggregate(agg_query)

    distinct_counts = {x["_id"]: x["count"] for x in distinct_counts_cur}

    if None in distinct_counts:
        del distinct_counts[None]

    return distinct_counts

save_one

save_one(snapshot: dict, ctime: datetime)

Saves snapshot to specified entity of current master document.

Will move snapshot to oversized snapshots if the maintained bucket is too large.

Source code in dp3/database/snapshots.py

def save_one(self, snapshot: dict, ctime: datetime):
    """Saves snapshot to specified entity of current master document.

    Will move snapshot to oversized snapshots if the maintained bucket is too large.
    """
    if "eid" not in snapshot:
        self.log.error("Snapshot is missing 'eid' field: %s", snapshot)
        return
    eid = snapshot["eid"]
    snapshot["_time_created"] = ctime

    snapshot_col = self._col()
    os_col = self._os_col()

    # Find out if the snapshot is oversized
    normal, oversized = self._get_state({eid})
    if normal:
        try:
            res = snapshot_col.update_one(
                self._filter_from_eid(eid) | {"count": {"$lt": self._snapshot_bucket_size}},
                {
                    "$set": {"last": snapshot},
                    "$push": {"history": {"$each": [snapshot], "$position": 0}},
                    "$inc": {"count": 1},
                    "$setOnInsert": {
                        "_id": self._bucket_id(eid, ctime),
                        "_time_created": ctime,
                        "oversized": False,
                        "latest": True,
                    },
                },
                upsert=True,
            )

            if res.upserted_id is not None:
                snapshot_col.update_many(
                    self._filter_from_eid(eid)
                    | {"latest": True, "count": self._snapshot_bucket_size},
                    {"$unset": {"latest": 1}},
                )
        except (WriteError, OperationFailure, DocumentTooLarge) as e:
            if e.code != BSON_OBJECT_TOO_LARGE:
                raise e
            # The snapshot is too large, move it to oversized snapshots
            self.log.info(f"Snapshot of {eid} is too large: {e}, marking as oversized.")
            self._migrate_to_oversized(eid, snapshot)
        except Exception as e:
            raise SnapshotCollectionError(
                f"Insert of snapshot {eid} failed: {e}, {snapshot}"
            ) from e
        return
    elif oversized:
        # Snapshot is already marked as oversized
        snapshot_col.update_one(
            self._filter_from_eid(eid),
            {"$set": {"last": snapshot, "_time_created": ctime}},
        )
        os_col.insert_one(snapshot)
        return

save_many

save_many(snapshots: list[dict], ctime: datetime)

Saves a list of snapshots of current master documents.

All snapshots must belong to same entity type.

Will move snapshots to oversized snapshots if the maintained bucket is too large. For better understanding, see save().

Source code in dp3/database/snapshots.py

def save_many(self, snapshots: list[dict], ctime: datetime):
    """
    Saves a list of snapshots of current master documents.

    All snapshots must belong to same entity type.

    Will move snapshots to oversized snapshots if the maintained bucket is too large.
    For better understanding, see `save()`.
    """

    for snapshot in snapshots:
        snapshot["_time_created"] = ctime

    snapshot_col = self._col()
    os_col = self._os_col()

    snapshots_by_eid = defaultdict(list)
    for snapshot in snapshots:
        if "eid" not in snapshot:
            continue
        if "_id" in snapshot:
            del snapshot["_id"]
        snapshots_by_eid[snapshot["eid"]].append(snapshot)

    # Find out if any of the snapshots are oversized
    normal, oversized = self._get_state(set(snapshots_by_eid.keys()))

    upserts = []
    update_originals: list[list[dict]] = []
    oversized_inserts = []
    oversized_updates = []

    # A normal snapshot, shift the last snapshot to history and update last
    for eid in normal:
        eid_snapshots = snapshots_by_eid[eid]
        query = self._filter_from_eid(eid) | {"count": {"$lt": self._snapshot_bucket_size}}
        update = {
            "$set": {"last": eid_snapshots[-1]},
            "$push": {"history": {"$each": eid_snapshots, "$position": 0}},
            "$inc": {"count": len(eid_snapshots)},
            "$setOnInsert": {
                "_id": self._bucket_id(eid, ctime),
                "_time_created": ctime,
                "oversized": False,
                "latest": True,
            },
        }
        update_statement = {"q": query, "u": update, "multi": False, "upsert": True}
        update_size = len(BSON.encode(update_statement, codec_options=self._db.codec_options))
        if update_size > self._max_bson_size():
            self.log.info(
                "Snapshot update for %s is too large (%d bytes), marking as oversized.",
                eid,
                update_size,
            )
            self._migrate_snapshot_batch(eid_snapshots, oversized_inserts)
            continue

        upserts.append(UpdateOne(query, update, upsert=True))
        update_originals.append(eid_snapshots)

    # Snapshot is already marked as oversized
    for eid in oversized:
        oversized_inserts.extend(snapshots_by_eid[eid])
        oversized_updates.append(
            UpdateOne(
                self._filter_from_eid(eid),
                {
                    "$set": {
                        "last": snapshots_by_eid[eid][-1],
                        "_time_created": ctime,
                    }
                },
            )
        )

    if upserts:
        try:
            processed_count, upserted_ids = self._write_snapshot_upserts(
                snapshot_col, upserts, update_originals, oversized_inserts
            )

            # Unset latest snapshots if new snapshots were inserted
            if upserted_ids:
                unset_latest_updates = [
                    UpdateMany(
                        self._filter_from_bid(upsert_id)
                        | {"latest": True, "count": self._snapshot_bucket_size},
                        {"$unset": {"latest": 1}},
                    )
                    for upsert_id in upserted_ids
                ]
                up_res = snapshot_col.bulk_write(unset_latest_updates)
                if up_res.modified_count != len(upserted_ids):
                    self.log.info(
                        "Upserted the first snapshot for %d entities.",
                        len(upserted_ids) - up_res.modified_count,
                    )

            if processed_count != len(upserts):
                self.log.error(
                    "Some snapshots were not updated, %s != %s",
                    processed_count,
                    len(upserts),
                )
        except (BulkWriteError, OperationFailure, SnapshotCollectionError):
            raise
        except Exception as e:
            raise SnapshotCollectionError(f"Upsert of snapshots failed: {str(e)[:2048]}") from e

    # Update the oversized snapshots
    if oversized_inserts:
        try:
            if oversized_updates:
                snapshot_col.bulk_write(oversized_updates)
            os_col.insert_many(oversized_inserts)
        except Exception as e:
            raise SnapshotCollectionError(f"Insert of snapshots failed: {str(e)[:2048]}") from e

delete_old

delete_old(t_old: datetime) -> int

Delete old snapshots.

Periodically called from HistoryManager.

Source code in dp3/database/snapshots.py

def delete_old(self, t_old: datetime) -> int:
    """Delete old snapshots.

    Periodically called from HistoryManager.
    """
    deleted = 0
    try:
        res = self._col().delete_many(
            {"_time_created": {"$lte": t_old - self._bucket_delta}},
        )
        deleted += res.deleted_count * self._snapshot_bucket_size
        res = self._os_col().delete_many({"_time_created": {"$lt": t_old}})
        deleted += res.deleted_count
    except Exception as e:
        raise SnapshotCollectionError(f"Delete of olds snapshots failed: {e}") from e
    return deleted

delete_eid

delete_eid(eid: AnyEidT) -> int

Delete all snapshots of eid.

Source code in dp3/database/snapshots.py

def delete_eid(self, eid: AnyEidT) -> int:
    """Delete all snapshots of `eid`."""
    try:
        res = self._col().delete_many(self._filter_from_eid(eid))
        del_cnt = res.deleted_count * self._snapshot_bucket_size
        self.log.debug("deleted %s snapshots of %s/%s.", del_cnt, self.entity_type, eid)

        res = self._os_col().delete_many({"eid": eid})
        self.log.debug(
            "Deleted %s oversized snapshots of %s/%s.", res.deleted_count, self.entity_type, eid
        )
        return del_cnt + res.deleted_count
    except Exception as e:
        raise SnapshotCollectionError(f"Delete of failed: {e}\n{eid}") from e
    finally:
        self._invalidate_snapshot_state({eid})

delete_eids

delete_eids(eids: list[Any]) -> int

Delete all snapshots of eids.

Source code in dp3/database/snapshots.py

def delete_eids(self, eids: list[Any]) -> int:
    """Delete all snapshots of `eids`."""
    try:
        res = self._col().delete_many(self._filter_from_eids(eids))
        del_cnt = res.deleted_count * self._snapshot_bucket_size
        self.log.debug("Deleted %s snapshots of %s (%s).", del_cnt, self.entity_type, len(eids))
        res = self._os_col().delete_many({"eid": {"$in": eids}})
        self.log.debug(
            "Deleted %s oversized snapshots of %s (%s).",
            res.deleted_count,
            self.entity_type,
            len(eids),
        )
        return del_cnt + res.deleted_count
    except Exception as e:
        raise SnapshotCollectionError(f"Delete of snapshots failed: {e}\n{eids}") from e
    finally:
        self._invalidate_snapshot_state(eids)

BinaryEidSnapshots

BinaryEidSnapshots(db: Database, entity_type: str, db_config: MongoConfig, model_spec: ModelSpec, snapshots_config: dict)

Bases: TypedSnapshotCollection, ABC

Source code in dp3/database/snapshots.py

def __init__(
    self,
    db: Database,
    entity_type: str,
    db_config: MongoConfig,
    model_spec: ModelSpec,
    snapshots_config: dict,
):
    self._db = db.with_options(codec_options=get_codec_options())

    if entity_type not in model_spec.entities:
        raise ValueError(f"Entity type '{entity_type}' not found in model spec")
    self.entity_type = entity_type
    self._col_name = f"{entity_type}#snapshots"
    self._os_col_name = f"{entity_type}#snapshots_oversized"

    self.attr_specs: dict[str, AttrSpecType] = model_spec.entity_attributes[entity_type]

    self.log = logging.getLogger(f"EntityDatabase.SnapshotCollection[{entity_type}]")

    self._normal_snapshot_eids = set()
    self._oversized_snapshot_eids = set()
    self._snapshot_bucket_size = db_config.storage.snapshot_bucket_size
    self._bucket_delta = self._get_snapshot_bucket_delta(snapshots_config)
    self._server_max_bson_size: int | None = None

SnapshotCollectionContainer

SnapshotCollectionContainer(db: Database, db_config: MongoConfig, model_spec: ModelSpec, snapshots_config: dict)

Container for all required snapshot collections, exposing the public interface.

Source code in dp3/database/snapshots.py

def __init__(
    self, db: Database, db_config: MongoConfig, model_spec: ModelSpec, snapshots_config: dict
):
    self._snapshot_collections = {}
    for entity_type, entity_spec in model_spec.entities.items():
        eid_type_name = entity_spec.id_data_type.root
        typed_collection: type[TypedSnapshotCollection] = entity_type2collection[eid_type_name]
        self._snapshot_collections[entity_type] = typed_collection(
            db, entity_type, db_config, model_spec, snapshots_config
        )

    self.log = logging.getLogger("EntityDatabase.SnapshotCollections")

save_one

save_one(entity_type: str, snapshot: dict, ctime: datetime) -> None

Save snapshot to specified entity of current master document.

Source code in dp3/database/snapshots.py

def save_one(self, entity_type: str, snapshot: dict, ctime: datetime) -> None:
    """Save snapshot to specified entity of current master document."""
    return self[entity_type].save_one(snapshot, ctime)

save_many

save_many(entity_type: str, snapshots: list[dict], ctime: datetime) -> None

Saves a list of snapshots of current master documents.

All snapshots must belong to same entity type.

Source code in dp3/database/snapshots.py

def save_many(self, entity_type: str, snapshots: list[dict], ctime: datetime) -> None:
    """
    Saves a list of snapshots of current master documents.

    All snapshots must belong to same entity type.
    """
    return self[entity_type].save_many(snapshots, ctime)

get_latest_one

get_latest_one(entity_type: str, eid: AnyEidT) -> dict

Get latest snapshot of given eid.

If doesn't exist, returns {}.

Source code in dp3/database/snapshots.py

def get_latest_one(self, entity_type: str, eid: AnyEidT) -> dict:
    """Get latest snapshot of given eid.

    If doesn't exist, returns {}.
    """
    return self[entity_type].get_latest_one(eid)

find_latest

find_latest(entity_type: str, fulltext_filters: dict[str, str] | None = None, generic_filter: dict[str, Any] | None = None) -> Cursor

Find latest snapshots of given etype.

This method is useful for displaying data on web.

Returns only documents matching generic_filter and fulltext_filters (dictionary attribute - fulltext filter). Fulltext filters are interpreted as regular expressions. Only string values may be filtered this way. There's no validation that queried attribute can be fulltext filtered. Only plain and observation attributes with string-based data types can be queried. Array and set data types are supported as well as long as they are not multi value at the same time. If you need to filter EIDs, ensure the EID is string, then use attribute eid. Otherwise, use generic filter.

Generic filter allows filtering using generic MongoDB query (including $and, $or, $lt, etc.). For querying non-JSON-native types, you can use magic strings, such as "$$IPv4{<ip address>}" for IPv4 addresses. The full spec with examples is in the magic strings module.

Generic and fulltext filters are merged - fulltext overrides conflicting keys.

May raise SnapshotCollectionError if query is invalid.

Returns only documents matching generic_filter and fulltext_filters, does not count them.

Source code in dp3/database/snapshots.py

def find_latest(
    self,
    entity_type: str,
    fulltext_filters: dict[str, str] | None = None,
    generic_filter: dict[str, Any] | None = None,
) -> Cursor:
    """Find latest snapshots of given `etype`.

    This method is useful for displaying data on web.

    Returns only documents matching `generic_filter` and `fulltext_filters`
    (dictionary attribute - fulltext filter).
    Fulltext filters are interpreted as regular expressions.
    Only string values may be filtered this way. There's no validation that queried attribute
    can be fulltext filtered.
    Only plain and observation attributes with string-based data types can be queried.
    Array and set data types are supported as well as long as they are not multi value
    at the same time.
    If you need to filter EIDs, ensure the EID is string, then use attribute `eid`.
    Otherwise, use generic filter.

    Generic filter allows filtering using generic MongoDB query (including `$and`, `$or`,
    `$lt`, etc.).
    For querying non-JSON-native types, you can use magic strings, such as
    `"$$IPv4{<ip address>}"` for IPv4 addresses. The full spec with examples is in the
    [magic strings module][dp3.database.magic].

    Generic and fulltext filters are merged - fulltext overrides conflicting keys.

    May raise `SnapshotCollectionError` if query is invalid.

    Returns only documents matching `generic_filter` and `fulltext_filters`,
    does not count them.
    """
    return self[entity_type].find_latest(fulltext_filters, generic_filter)

count_latest

count_latest(entity_type: str, fulltext_filters: dict[str, str] | None = None, generic_filter: dict[str, Any] | None = None) -> int

Count latest snapshots of given etype.

see find_latest for more information.

Returns only count of documents matching generic_filter and fulltext_filters.

Note that this method may take much longer than find_latest on larger databases, as it does count all documents, not just return the first few.

Source code in dp3/database/snapshots.py

def count_latest(
    self,
    entity_type: str,
    fulltext_filters: dict[str, str] | None = None,
    generic_filter: dict[str, Any] | None = None,
) -> int:
    """Count latest snapshots of given `etype`.

    see [`find_latest`][dp3.database.snapshots.SnapshotCollectionContainer.find_latest]
    for more information.

    Returns only count of documents matching `generic_filter` and `fulltext_filters`.

    Note that this method may take much longer than `find_latest` on larger databases,
    as it does count all documents, not just return the first few.
    """
    return self[entity_type].count_latest(fulltext_filters, generic_filter)

get_by_eid

get_by_eid(entity_type: str, eid: AnyEidT, t1: datetime | None = None, t2: datetime | None = None, skip: int = 0, limit: int = 0) -> Cursor | CommandCursor

Get all (or filtered) snapshots of given eid.

This method is useful for displaying eid's history on web.

Parameters:

Name	Type	Description	Default
entity_type	str	name of entity type	required
eid	AnyEidT	id of entity, to which data-points correspond	required
t1	datetime | None	left value of time interval (inclusive)	None
t2	datetime | None	right value of time interval (inclusive)	None
skip	int	number of snapshots to skip	0
limit	int	max number of snapshots to return, 0 means no limit	0

Source code in dp3/database/snapshots.py

def get_by_eid(
    self,
    entity_type: str,
    eid: AnyEidT,
    t1: datetime | None = None,
    t2: datetime | None = None,
    skip: int = 0,
    limit: int = 0,
) -> Cursor | CommandCursor:
    """Get all (or filtered) snapshots of given `eid`.

    This method is useful for displaying `eid`'s history on web.

    Args:
        entity_type: name of entity type
        eid: id of entity, to which data-points correspond
        t1: left value of time interval (inclusive)
        t2: right value of time interval (inclusive)
        skip: number of snapshots to skip
        limit: max number of snapshots to return, `0` means no limit
    """
    return self[entity_type].get_by_eid(eid, t1, t2, skip, limit)

get_distinct_val_count

get_distinct_val_count(entity_type: str, attr: str) -> dict[Any, int]

Counts occurrences of distinct values of given attribute in snapshots.

Returns dictionary mapping value -> count.

Works for all plain and observation data types except dict and json.

Source code in dp3/database/snapshots.py

def get_distinct_val_count(self, entity_type: str, attr: str) -> dict[Any, int]:
    """Counts occurrences of distinct values of given attribute in snapshots.

    Returns dictionary mapping value -> count.

    Works for all plain and observation data types except `dict` and `json`.
    """
    return self[entity_type].get_distinct_val_count(attr)

delete_old

delete_old(t_old: datetime) -> int

Delete old snapshots, may raise SnapshotCollectionError.

Periodically called from HistoryManager. Returns: number of deleted snapshots.

Source code in dp3/database/snapshots.py

def delete_old(self, t_old: datetime) -> int:
    """Delete old snapshots, may raise `SnapshotCollectionError`.

    Periodically called from HistoryManager.
    Returns:
         number of deleted snapshots.
    """
    deleted_total = 0
    for collection in self._snapshot_collections.values():
        try:
            deleted_total += collection.delete_old(t_old)
        except Exception as e:
            raise SnapshotCollectionError(f"Delete of old snapshots failed: {e}") from e
    return deleted_total

delete_eid

delete_eid(entity_type: str, eid: AnyEidT) -> int

Delete snapshots of given eids.

Source code in dp3/database/snapshots.py

def delete_eid(self, entity_type: str, eid: AnyEidT) -> int:
    """Delete snapshots of given `eids`."""
    return self[entity_type].delete_eid(eid)

delete_eids

delete_eids(entity_type: str, eids: list[Any]) -> int

Delete snapshots of given eids.

Source code in dp3/database/snapshots.py

def delete_eids(self, entity_type: str, eids: list[Any]) -> int:
    """Delete snapshots of given `eids`."""
    return self[entity_type].delete_eids(eids)