Skip to content

Service Order Notes

ServiceOrderNotes(perfdb)

Class used for handling Service Order Note data. Can be accessed via perfdb.service_orders.notes.

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(service_order_names=None, service_order_sap_ids=None, note_descriptions=None, note_sap_ids=None, statuses=None, filter_type='and')

Deletes service order notes from the database.

Be cautious when using this method, as it will permanently delete data.

Parameters:

  • service_order_names

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

    List of service order names to retrieve information for.

    If None, retrieves information for all available. By default None.

  • note_descriptions

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

    List of service order note descriptions to retrieve information for. Will be treated as regex.

    If None, retrieves information for all available. By default None.

  • note_sap_ids

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

    List of service order note SAP IDs to retrieve information for.

    If None, retrieves information for all available. By default None.

  • statuses

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

    List of statuses to filter the notes by.

    If None, retrieves information for all available. By default None.

  • filter_type

    (Literal['and', 'or'], default: 'and' ) –

    How to treat multiple filters. Can be one of ["and", "or"]

    By default "and"

Source code in echo_postgres/service_order_notes.py 
@validate_call
def delete(
    self,
    service_order_names: list[str] | None = None,
    service_order_sap_ids: list[int] | None = None,
    note_descriptions: list[str] | None = None,
    note_sap_ids: list[int] | None = None,
    statuses: list[str] | None = None,
    filter_type: Literal["and", "or"] = "and",
) -> None:
    """Deletes service order notes from the database.

    Be cautious when using this method, as it will permanently delete data.

    Parameters
    ----------
    service_order_names : list[str] | None, optional
        List of service order names to retrieve information for.

        If None, retrieves information for all available. By default None.
    note_descriptions : list[str] | None, optional
        List of service order note descriptions to retrieve information for. Will be treated as regex.

        If None, retrieves information for all available. By default None.
    note_sap_ids : list[int] | None, optional
        List of service order note SAP IDs to retrieve information for.

        If None, retrieves information for all available. By default None.
    statuses : list[str] | None, optional
        List of statuses to filter the notes by.

        If None, retrieves information for all available. By default None.
    filter_type : Literal["and", "or"], optional
        How to treat multiple filters. Can be one of ["and", "or"]

        By default "and"
    """
    # checking if at least one filter is provided
    if not any(
        [
            service_order_names,
            service_order_sap_ids,
            note_descriptions,
            note_sap_ids,
            statuses,
        ],
    ):
        raise ValueError("At least one filter must be provided to delete service order notes.")

    # getting ids to delete
    ids = self.get_ids(
        service_order_names=service_order_names,
        service_order_sap_ids=service_order_sap_ids,
        note_descriptions=note_descriptions,
        note_sap_ids=note_sap_ids,
        statuses=statuses,
        filter_type=filter_type,
    )

    if not ids:
        logger.debug("No service order notes found to delete based on the provided filters.")
        return

    query = sql.SQL(
        "DELETE FROM performance.service_order_notes WHERE id = ANY({ids})",
    ).format(
        ids=sql.Literal(list(ids.values())),
    )

    with self._perfdb.conn.reconnect() as conn:
        result = conn.execute(query)

    rows = result.rowcount if result else 0
    logger.debug(f"Deleted {rows} service order notes from the database.")

get(service_order_names=None, service_order_sap_ids=None, note_descriptions=None, note_sap_ids=None, statuses=None, filter_type='and', output_type='dict')

Retrieves service order note information from the database.

The most useful keys/columns returned are:

  • service_order_id
  • service_order_name
  • service_order_sap_id
  • note_id
  • note_sap_id (is the index if output_type is "DataFrame")
  • note_description
  • status_id
  • status_name
  • status_display_name
  • status_sap_name

Parameters:

  • service_order_names

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

    List of service order names to retrieve information for.

    If None, retrieves information for all available. By default None.

  • service_order_sap_ids

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

    List of service order SAP IDs to retrieve information for.

    If None, retrieves information for all available. By default None.

  • note_descriptions

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

    List of service order note descriptions to retrieve information for. Will be treated as regex.

    If None, retrieves information for all available. By default None.

  • note_sap_ids

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

    List of service order note SAP IDs to retrieve information for.

    If None, retrieves information for all available. By default None.

  • statuses

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

    List of statuses to filter the notes by.

    If None, retrieves information for all available. By default None.

  • filter_type

    (Literal['and', 'or'], default: 'and' ) –

    How to treat multiple filters. Can be one of ["and", "or"]

    By default "and"

  • output_type

    (Literal['dict', 'DataFrame', 'pl.DataFrame'], default: 'dict' ) –

    Output type of the data. Can be one of ["dict", "DataFrame", "pl.DataFrame"] By default "dict"

Returns:

  • dict[str, dict[str, Any]] –

    In case output_type is "dict", returns a dictionary in the format {note_sap_id: {attribute: value, ...}, ...}

  • DataFrame –

    In case output_type is "DataFrame", returns a DataFrame with the following format: index = note_sap_id, columns = [attribute, ...]

  • DataFrame –

    In case output_type is "pl.DataFrame", returns a Polars DataFrame

Source code in echo_postgres/service_order_notes.py 
@validate_call
def get(
    self,
    service_order_names: list[str] | None = None,
    service_order_sap_ids: list[int] | None = None,
    note_descriptions: list[str] | None = None,
    note_sap_ids: list[int] | None = None,
    statuses: list[str] | None = None,
    filter_type: Literal["and", "or"] = "and",
    output_type: Literal["dict", "DataFrame", "pl.DataFrame"] = "dict",
) -> dict[str, dict[str, Any]] | pd.DataFrame | pl.DataFrame:
    """Retrieves service order note information from the database.

    The most useful keys/columns returned are:

    - service_order_id
    - service_order_name
    - service_order_sap_id
    - note_id
    - note_sap_id (is the index if output_type is "DataFrame")
    - note_description
    - status_id
    - status_name
    - status_display_name
    - status_sap_name

    Parameters
    ----------
    service_order_names : list[str] | None, optional
        List of service order names to retrieve information for.


        If None, retrieves information for all available. By default None.
    service_order_sap_ids : list[int] | None, optional
        List of service order SAP IDs to retrieve information for.

        If None, retrieves information for all available. By default None.
    note_descriptions : list[str] | None, optional
        List of service order note descriptions to retrieve information for. Will be treated as regex.

        If None, retrieves information for all available. By default None.
    note_sap_ids : list[int] | None, optional
        List of service order note SAP IDs to retrieve information for.

        If None, retrieves information for all available. By default None.
    statuses : list[str] | None, optional
        List of statuses to filter the notes by.

        If None, retrieves information for all available. By default None.
    filter_type : Literal["and", "or"], optional
        How to treat multiple filters. Can be one of ["and", "or"]

        By default "and"
    output_type : Literal["dict", "DataFrame", "pl.DataFrame"], optional
        Output type of the data. Can be one of ["dict", "DataFrame", "pl.DataFrame"]
        By default "dict"

    Returns
    -------
    dict[str, dict[str, Any]]
        In case output_type is "dict", returns a dictionary in the format {note_sap_id: {attribute: value, ...}, ...}
    pd.DataFrame
        In case output_type is "DataFrame", returns a DataFrame with the following format: index = note_sap_id, columns = [attribute, ...]
    pl.DataFrame
        In case output_type is "pl.DataFrame", returns a Polars DataFrame
    """
    where_query = self._check_get_args(
        service_order_names=service_order_names,
        service_order_sap_ids=service_order_sap_ids,
        note_descriptions=note_descriptions,
        note_sap_ids=note_sap_ids,
        statuses=statuses,
        filter_type=filter_type,
    )

    query = sql.SQL(
        "SELECT * FROM performance.v_service_order_notes {where_query} ORDER BY service_order_name, note_sap_id",
    ).format(
        where_query=where_query,
    )

    with self._perfdb.conn.reconnect() as conn:
        df = conn.read_to_polars(query)

    if output_type == "pl.DataFrame":
        return df

    df = df.to_pandas(use_pyarrow_extension_array=True)
    df = df.set_index("note_sap_id")

    if output_type == "DataFrame":
        return df

    return df.to_dict(orient="index")

get_ids(service_order_names=None, service_order_sap_ids=None, note_descriptions=None, note_sap_ids=None, statuses=None, filter_type='and')

Retrieves service order note ids from the database.

Parameters:

  • service_order_names

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

    List of service order names to retrieve information for.

    If None, retrieves information for all available. By default None.

  • note_descriptions

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

    List of service order note descriptions to retrieve information for. Will be treated as regex.

    If None, retrieves information for all available. By default None.

  • note_sap_ids

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

    List of service order note SAP IDs to retrieve information for.

    If None, retrieves information for all available. By default None.

  • statuses

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

    List of statuses to filter the notes by.

    If None, retrieves information for all available. By default None.

  • filter_type

    (Literal['and', 'or'], default: 'and' ) –

    How to treat multiple filters. Can be one of ["and", "or"]

    By default "and"

Returns:

  • dict[str, int] –

    Dictionary in the format {note_sap_id: note_id, ...}

Source code in echo_postgres/service_order_notes.py 
@validate_call
def get_ids(
    self,
    service_order_names: list[str] | None = None,
    service_order_sap_ids: list[int] | None = None,
    note_descriptions: list[str] | None = None,
    note_sap_ids: list[int] | None = None,
    statuses: list[str] | None = None,
    filter_type: Literal["and", "or"] = "and",
) -> dict[str, int]:
    """Retrieves service order note ids from the database.

    Parameters
    ----------
    service_order_names : list[str] | None, optional
        List of service order names to retrieve information for.

        If None, retrieves information for all available. By default None.
    note_descriptions : list[str] | None, optional
        List of service order note descriptions to retrieve information for. Will be treated as regex.

        If None, retrieves information for all available. By default None.
    note_sap_ids : list[int] | None, optional
        List of service order note SAP IDs to retrieve information for.

        If None, retrieves information for all available. By default None.
    statuses : list[str] | None, optional
        List of statuses to filter the notes by.

        If None, retrieves information for all available. By default None.
    filter_type : Literal["and", "or"], optional
        How to treat multiple filters. Can be one of ["and", "or"]

        By default "and"

    Returns
    -------
    dict[str, int]
        Dictionary in the format {note_sap_id: note_id, ...}
    """
    where = self._check_get_args(
        service_order_names=service_order_names,
        service_order_sap_ids=service_order_sap_ids,
        note_descriptions=note_descriptions,
        note_sap_ids=note_sap_ids,
        statuses=statuses,
        filter_type=filter_type,
    )

    query = sql.SQL(
        "SELECT note_sap_id, note_id FROM performance.v_service_order_notes {where_query} ORDER BY note_sap_id",
    ).format(
        where_query=where,
    )

    with self._perfdb.conn.reconnect() as conn:
        df = conn.read_to_polars(query)

    return dict(zip(df["note_sap_id"], df["note_id"], strict=False))

insert(description, service_order_name, sap_id, on_conflict='ignore')

Inserts a new service order note, linked with an existing service order id.

Parameters:

  • description

    (str) –

    Description of the service order note to be inserted.

  • service_order_name

    (str) –

    Name of the service order to link the note to. Must already exist in the database.

  • sap_id

    (int) –

    SAP ID of the service order note to be inserted.

  • on_conflict

    (Literal['ignore', 'update'], default: 'ignore' ) –

    Action to take in case of conflict. Can be one of:

    • "ignore": ignores the new data if a conflict occurs (default)
    • "update": updates the existing data with the new data in case of conflict

    The conflict will be determined based on the sap_id of the service order note.

Returns:

  • int | None –

    ID of the inserted service order note. Returns None if on_conflict is "ignore" and a conflict occurs.

Source code in echo_postgres/service_order_notes.py 
@validate_call
def insert(
    self,
    description: str,
    service_order_name: str,
    sap_id: int,
    on_conflict: Literal["ignore", "update"] = "ignore",
) -> int | None:
    """Inserts a new service order note, linked with an existing service order id.

    Parameters
    ----------
    description : str
        Description of the service order note to be inserted.
    service_order_name : str
        Name of the service order to link the note to. Must already exist in the database.
    sap_id : int
        SAP ID of the service order note to be inserted.
    on_conflict : Literal["ignore", "update"], optional
        Action to take in case of conflict. Can be one of:

        - "ignore": ignores the new data if a conflict occurs (default)
        - "update": updates the existing data with the new data in case of conflict

        The conflict will be determined based on the sap_id of the service order note.

    Returns
    -------
    int | None
        ID of the inserted service order note. Returns None if on_conflict is "ignore" and a conflict occurs.
    """
    # getting service order id
    service_order_ids = self._perfdb.service_orders.get_ids(
        names=[service_order_name],
    )
    if not service_order_ids:
        raise ValueError(f"Service order with name '{service_order_name}' does not exist in the database.")

    service_order_id = service_order_ids[service_order_name]

    query = sql.SQL(
        """
        INSERT INTO performance.service_order_notes (description, service_order_id, sap_id)
        VALUES ({note_description}, {service_order_id}, {note_sap_id})
        ON CONFLICT (sap_id) DO {conflict_action}
        RETURNING id
        """,
    ).format(
        note_description=sql.Literal(description),
        service_order_id=sql.Literal(service_order_id),
        note_sap_id=sql.Literal(sap_id),
        conflict_action=sql.SQL(
            "NOTHING"
            if on_conflict == "ignore"
            else "UPDATE SET description = EXCLUDED.description, service_order_id = EXCLUDED.service_order_id",
        ),
    )

    with self._perfdb.conn.reconnect() as conn:
        result = conn.execute(query)

    if result and result.rowcount == 1:
        inserted_id = result.fetchone()[0]
        return inserted_id

    return None

update(note_id, description=None, service_order_name=None, sap_id=None)

Updates an existing service order note in the database.

Parameters:

  • note_id

    (int) –

    ID of the service order note to be updated.

  • description

    (str | None, default: None ) –

    New description for the service order note. If None, the description will not be updated.

  • service_order_name

    (str | None, default: None ) –

    New service order name to link the note to. If None, the service order will not be updated.

  • sap_id

    (int | None, default: None ) –

    New SAP ID for the service order note. If None, the SAP ID will not be updated.

Source code in echo_postgres/service_order_notes.py 
@validate_call
def update(
    self,
    note_id: int,
    description: str | None = None,
    service_order_name: str | None = None,
    sap_id: int | None = None,
) -> None:
    """Updates an existing service order note in the database.

    Parameters
    ----------
    note_id : int
        ID of the service order note to be updated.
    description : str | None, optional
        New description for the service order note. If None, the description will not be updated.
    service_order_name : str | None, optional
        New service order name to link the note to. If None, the service order will not be updated.
    sap_id : int | None, optional
        New SAP ID for the service order note. If None, the SAP ID will not be updated.
    """
    # check if at least one field to update is provided
    if not any([description, service_order_name, sap_id]):
        raise ValueError("At least one field to update must be provided.")

    set_clauses = []
    if description is not None:
        set_clauses.append(
            sql.SQL("description = {note_description}").format(
                note_description=sql.Literal(description),
            ),
        )
    if service_order_name is not None:
        # getting service order id
        service_order_ids = self._perfdb.service_orders.get_ids(
            names=[service_order_name],
        )
        if not service_order_ids:
            raise ValueError(f"Service order with name '{service_order_name}' does not exist in the database.")

        service_order_id = service_order_ids[service_order_name]

        set_clauses.append(
            sql.SQL("service_order_id = {service_order_id}").format(
                service_order_id=sql.Literal(service_order_id),
            ),
        )
    if sap_id is not None:
        set_clauses.append(
            sql.SQL("sap_id = {note_sap_id}").format(
                note_sap_id=sql.Literal(sap_id),
            ),
        )

    query = sql.SQL(
        """
        UPDATE performance.service_order_notes
        SET {set_clauses}
        WHERE id = {note_id}
        """,
    ).format(
        set_clauses=sql.SQL(", ").join(set_clauses),
        note_id=sql.Literal(note_id),
    )

    with self._perfdb.conn.reconnect() as conn:
        result = conn.execute(query)

    rows = result.rowcount if result else 0
    if rows == 0:
        logger.warning(f"No service order note with id {id} was found to update.")
    else:
        logger.debug(f"Updated {rows} service order note with id {id}.")