Document Instances - Events

DocumentInstanceEvents(perfdb)

Class used for handling document events. Can be accessed via perfdb.documents.instances.events.

Parameters:

• perfdb

  (PerfDB) –

  Top level object carrying all functionality and the connection handler.

Source code in echo_postgres/perfdb_root.py

def __init__(self, perfdb: e_pg.PerfDB) -> None:
    """Base class that all subclasses should inherit from.

    Parameters
    ----------
    perfdb : PerfDB
        Top level object carrying all functionality and the connection handler.

    """
    self._perfdb: e_pg.PerfDB = perfdb

delete(document_id, event_ids=None)

Deletes events for a given document.

Parameters:

• document_id

  (int) –

  Document id

• event_ids

  (list[int] | None, default: None ) –

  List of events ids to be deleted. If None all will be deleted. By default, None.

Source code in echo_postgres/document_instance_events.py

@validate_call
def delete(self, document_id: int, event_ids: list[int] | None = None) -> None:
    """Deletes events for a given document.

    Parameters
    ----------
    document_id : int
        Document id
    event_ids : list[int] | None, optional
        List of events ids to be deleted. If None all will be deleted. By default, None.
    """
    query = sql.SQL("SELECT * FROM fn_delete_document_events({document_id}, {events})").format(
        document_id=sql.Literal(document_id),
        events=sql.SQL("NULL")
        if event_ids is None
        else sql.SQL("ARRAY[{event_ids}]").format(event_ids=sql.SQL(",").join(map(sql.Literal, event_ids))),
    )
    with self._perfdb.conn.reconnect() as conn:
        conn.execute(query, skip_retry=True)

insert(document_id, event_ids)

Inserts events for a given document.

Parameters:

• document_id

  (int) –

  Document id

• event_ids

  (list[int]) –

  List of event ids to be inserted

Source code in echo_postgres/document_instance_events.py

@validate_call
def insert(self, document_id: int, event_ids: list[int]) -> None:
    """Inserts events for a given document.

    Parameters
    ----------
    document_id : int
        Document id
    event_ids : list[int]
        List of event ids to be inserted
    """
    query = sql.SQL("SELECT * FROM fn_map_document_events({document_id}, ARRAY[{event_ids}])").format(
        document_id=sql.Literal(document_id),
        event_ids=sql.SQL(",").join(map(sql.Literal, event_ids)),
    )
    with self._perfdb.conn.reconnect() as conn:
        conn.execute(query, skip_retry=True)