Comments - Events

CommentsEvents(perfdb)

Class used for handling comments events. Can be accessed via perfdb.comments.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(comment_id, event_ids=None)

Deletes events for a given comment.

Parameters:

• comment_id

  (int) –

  comment 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/comments_events.py

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

    Parameters
    ----------
    comment_id : int
        comment id
    event_ids : list[int] | None, optional
        List of events ids to be deleted. If None all will be deleted. By default, None.
    """
    # checking arguments
    if isinstance(event_ids, list):
        if len(event_ids) == 0:
            raise ValueError("events must not be empty")
        if not all(isinstance(event, int) for event in event_ids):
            raise TypeError("All events must be strings")

    query = sql.SQL("SELECT * FROM fn_delete_comment_events({comment_id}, {events})").format(
        comment_id=sql.Literal(comment_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(comment_id, event_ids)

Inserts events for a given comment.

Parameters:

• comment_id

  (int) –

  comment id

• event_ids

  (list[int]) –

  List of event ids to be inserted

Source code in echo_postgres/comments_events.py

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

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