Comments - IV Curve Report Details

CommentsIVCurveReportDetails(perfdb)

Class used for handling comments events. Can be accessed via perfdb.comments.ivcurve.reports.details.

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, iv_curve_report_detail_ids=None)

Deletes comments associated with iv curve report details. This will delete both the mapping and the comment itself.

One of comment_id or iv_curve_report_detail_ids must be provided.

Parameters:

• comment_id

  (int | None) –

  Comment id to delete. If None, deletes all comments for the given iv curve report details. By default, None.

• iv_curve_report_detail_ids

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

  List of iv curve report detail ids to delete the comments for. If None all will be deleted. By default, None.

Source code in echo_postgres/comments_ivcurve_report_details.py

@validate_call
def delete(self, comment_id: int | None, iv_curve_report_detail_ids: list[int] | None = None) -> None:
    """Deletes comments associated with iv curve report details. This will delete both the mapping and the comment itself.

    One of comment_id or iv_curve_report_detail_ids must be provided.

    Parameters
    ----------
    comment_id : int | None, optional
        Comment id to delete. If None, deletes all comments for the given iv curve report details. By default, None.
    iv_curve_report_detail_ids : list[int] | None, optional
        List of iv curve report detail 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 iv_curve_report_detail_ids is None:
        raise ValueError("At least one of comment_id or iv_curve_report_detail_ids must be provided.")

    query = sql.SQL(
        "SELECT * FROM fn_delete_comment(_comment_id =>{comment_id}, _reference_ids => {iv_curve_report_detail_ids}, _reference_type => 'iv_curve_report_details')",
    ).format(
        comment_id=sql.Literal(comment_id) if comment_id is not None else sql.SQL("NULL"),
        iv_curve_report_detail_ids=sql.SQL("NULL")
        if iv_curve_report_detail_ids is None
        else sql.SQL("ARRAY[{iv_curve_report_detail_ids}]").format(
            iv_curve_report_detail_ids=sql.SQL(",").join(map(sql.Literal, iv_curve_report_detail_ids)),
        ),
    )
    with self._perfdb.conn.reconnect() as conn:
        conn.execute(query, skip_retry=True)

get_ids(iv_curve_report_detail_ids)

Gets the comment ids for a given list of iv curve report detail ids.

Parameters:

• iv_curve_report_detail_ids

  (list[int]) –

  List of iv curve report detail ids to get the comment for.

Returns:

• dict[int, list[int]] –

  Dictionary mapping iv curve report detail ids to comment ids.

Source code in echo_postgres/comments_ivcurve_report_details.py

@validate_call
def get_ids(self, iv_curve_report_detail_ids: list[int]) -> dict[int, list[int]]:
    """Gets the comment ids for a given list of iv curve report detail ids.

    Parameters
    ----------
    iv_curve_report_detail_ids : list[int]
        List of iv curve report detail ids to get the comment for.

    Returns
    -------
    dict[int, list[int]]
        Dictionary mapping iv curve report detail ids to comment ids.
    """
    if len(iv_curve_report_detail_ids) == 0:
        return []

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

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

    return result

insert(comment_id, iv_curve_report_detail_ids)

Inserts iv curve report details for a given comment.

Parameters:

• comment_id

  (int) –

  comment id

• iv_curve_report_detail_ids

  (list[int]) –

  List of iv curve report detail ids to be inserted

Source code in echo_postgres/comments_ivcurve_report_details.py

@validate_call
def insert(self, comment_id: int, iv_curve_report_detail_ids: list[int]) -> None:
    """Inserts iv curve report details for a given comment.

    Parameters
    ----------
    comment_id : int
        comment id
    iv_curve_report_detail_ids : list[int]
        List of iv curve report detail ids to be inserted
    """
    query = sql.SQL(
        "SELECT * FROM fn_map_comment(_comment_id => {comment_id}, _reference_ids => ARRAY[{iv_curve_report_detail_ids}], _reference_type => 'iv_curve_report_details')",
    ).format(
        comment_id=sql.Literal(comment_id),
        iv_curve_report_detail_ids=sql.SQL(",").join(map(sql.Literal, iv_curve_report_detail_ids)),
    )
    with self._perfdb.conn.reconnect() as conn:
        conn.execute(query, skip_retry=True)