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 comments associated with events. This will delete both the mapping and the comment itself.

One of comment_id or event_ids must be provided.

Parameters:

• comment_id

  (int | None) –

  Comment id to delete. If None, deletes all comments for the given events. By default, None.

• event_ids

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

  List of events ids to delete the comments for. 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 | None, event_ids: list[int] | None = None) -> None:
    """Deletes comments associated with events. This will delete both the mapping and the comment itself.

    One of comment_id or event_ids must be provided.

    Parameters
    ----------
    comment_id : int | None, optional
        Comment id to delete. If None, deletes all comments for the given events. By default, None.
    event_ids : list[int] | None, optional
        List of events ids to delete the comments for. If None all will be deleted. By default, None.
    """
    # validating if at least one parameter is provided
    if comment_id is None and event_ids is None:
        raise ValueError("At least one of comment_id or event_ids must be provided.")

    query = sql.SQL(
        "SELECT * FROM fn_delete_comment(_comment_id =>{comment_id}, _reference_ids => {events}, _reference_type => 'events')",
    ).format(
        comment_id=sql.Literal(comment_id) if comment_id is not None else sql.SQL("NULL"),
        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)

get_ids(event_ids)

Gets the comment event ids for a given list of event ids.

Parameters:

• event_ids

  (list[int]) –

  List of event ids to get the comment event ids for.

Returns:

• dict[int, list[int]] –

  Dictionary mapping event ids to comment ids.

Source code in echo_postgres/comments_events.py

@validate_call
def get_ids(self, event_ids: list[int]) -> dict[int, list[int]]:
    """Gets the comment event ids for a given list of event ids.

    Parameters
    ----------
    event_ids : list[int]
        List of event ids to get the comment event ids for.

    Returns
    -------
    dict[int, list[int]]
        Dictionary mapping event ids to comment ids.
    """
    if len(event_ids) == 0:
        return []

    query = sql.SQL("SELECT event_id, comment_id FROM performance.comment_events_mapping WHERE event_id IN ({event_ids})").format(
        event_ids=sql.SQL(",").join(map(sql.Literal, event_ids)),
    )
    with self._perfdb.conn.reconnect() as conn:
        result = conn.read_to_polars(query)

    # building the output dictionary
    grouped = result.group_by("event_id").agg(pl.col("comment_id").list())
    result = dict(zip(grouped["event_id"], grouped["comment_id"], strict=False))

    return result

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(_comment_id => {comment_id}, _reference_ids => ARRAY[{event_ids}], _reference_type => 'events')",
    ).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)