Service Order¶

`ServiceOrders(perfdb)` ¶

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

Parameters:

perfdb ¶
(PerfDB) –

Top level object carrying all functionality and the connection handler.

Source code in echo_postgres/service_orders.py

def __init__(self, perfdb: e_pg.PerfDB) -> None:
    """Class used for handling Service Order data. Can be accessed via `perfdb.service_orders`.

    Parameters
    ----------
    perfdb : PerfDB
        Top level object carrying all functionality and the connection handler.
    """
    super().__init__(perfdb)

    from .service_order_notes import ServiceOrderNotes
    from .service_order_status import ServiceOrderStatus

    # * subclasses

    self.notes = ServiceOrderNotes(perfdb)
    self.status = ServiceOrderStatus(perfdb)

`delete(names, descriptions=None, sap_ids=None, statuses=None, filter_type='and')` ¶

Deletes service orders from the database.

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

Parameters:

names ¶
(list[str] | None) –

List of names to retrieve information for.

If None, retrieves information for all available.
descriptions ¶
(list[str] | None, default: None ) –

List of descriptions (regex) to retrieve information for.

If None, retrieves information for all available.
sap_ids ¶
(list[int] | None, default: None ) –

List of SAP IDs to retrieve information for.

If None, retrieves information for all available.
statuses ¶
(list[str] | None, default: None ) –

List of statuses to retrieve information for.

If None, retrieves information for all available.
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_orders.py

@validate_call
def delete(
    self,
    names: list[str],
    descriptions: list[str] | None = None,
    sap_ids: list[int] | None = None,
    statuses: list[str] | None = None,
    filter_type: Literal["and", "or"] = "and",
) -> None:
    """Deletes service orders from the database.

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

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

        If None, retrieves information for all available.

    descriptions : list[str] | None, optional
        List of descriptions (regex) to retrieve information for.

        If None, retrieves information for all available.

    sap_ids : list[int] | None, optional
        List of SAP IDs to retrieve information for.

        If None, retrieves information for all available.

    statuses : list[str] | None, optional
        List of statuses to retrieve information for.

        If None, retrieves information for all available.

    filter_type : Literal["and", "or"], optional
        How to treat multiple filters. Can be one of ["and", "or"]. By default "and".
    """
    # check if at least one name filter is provided
    if not any([names, descriptions, sap_ids, statuses]):
        raise ValueError("At least one filter must be provided to delete service orders.")

    # getting ids to delete
    ids = self.get_ids(
        names=names,
        descriptions=descriptions,
        sap_ids=sap_ids,
        statuses=statuses,
        filter_type=filter_type,
    )

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

    query = sql.SQL(
        "DELETE FROM performance.service_orders 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 orders from the database.")

`get(names=None, descriptions=None, sap_ids=None, statuses=None, filter_type='and', output_type='dict')` ¶

Retrieves Service Order information from the database.

The most useful keys/columns returned are:

id
name (index if output_type is DataFrame)
sap_id
status_id
status_name
status_display_name
status_sap_name
description

Parameters:

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

List of names to retrieve information for.

If None, retrieves information for all available.
descriptions ¶
(list[str] | None, default: None ) –

List of descriptions (regex) to retrieve information for.

If None, retrieves information for all available.
sap_ids ¶
(list[int] | None, default: None ) –

List of SAP IDs to retrieve information for.

If None, retrieves information for all available.
statuses ¶
(list[str] | None, default: None ) –

List of statuses to retrieve information for.

If None, retrieves information for all available.
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 {name: {attribute: value, ...}, ...}
DataFrame –

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

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

Source code in echo_postgres/service_orders.py

@validate_call
def get(
    self,
    names: list[str] | None = None,
    descriptions: list[str] | None = None,
    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 information from the database.

    The most useful keys/columns returned are:

    - id
    - name (index if output_type is DataFrame)
    - sap_id
    - status_id
    - status_name
    - status_display_name
    - status_sap_name
    - description

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

        If None, retrieves information for all available.

    descriptions : list[str] | None, optional
        List of descriptions (regex) to retrieve information for.

        If None, retrieves information for all available.

    sap_ids : list[int] | None, optional
        List of SAP IDs to retrieve information for.

        If None, retrieves information for all available.

    statuses : list[str] | None, optional
        List of statuses to retrieve information for.

        If None, retrieves information for all available.

    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 {name: {attribute: value, ...}, ...}
    DataFrame
        In case output_type is "DataFrame", returns a DataFrame with the following format: index = name, columns = [attribute, ...]
    pl.DataFrame
        In case output_type is "pl.DataFrame", returns a Polars DataFrame
    """
    where_query = self._check_get_args(
        names=names,
        descriptions=descriptions,
        sap_ids=sap_ids,
        statuses=statuses,
        filter_type=filter_type,
    )

    query = sql.SQL(
        "SELECT * FROM performance.v_service_orders {where_query} ORDER BY name",
    ).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("name")

    if output_type == "DataFrame":
        return df

    return df.to_dict(orient="index")

`get_ids(names=None, descriptions=None, sap_ids=None, statuses=None, filter_type='and')` ¶

Retrieves service order ids from the database.

Parameters:

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

List of names to retrieve information for.

If None, retrieves information for all available.
descriptions ¶
(list[str] | None, default: None ) –

List of descriptions (regex) to retrieve information for.

If None, retrieves information for all available.
sap_ids ¶
(list[int] | None, default: None ) –

List of SAP IDs to retrieve information for.

If None, retrieves information for all available.
statuses ¶
(list[str] | None, default: None ) –

List of statuses to retrieve information for.

If None, retrieves information for all available.
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] –

A dictionary in the format {name: id, ...}

Source code in echo_postgres/service_orders.py

@validate_call
def get_ids(
    self,
    names: list[str] | None = None,
    descriptions: list[str] | None = None,
    sap_ids: list[int] | None = None,
    statuses: list[str] | None = None,
    filter_type: Literal["and", "or"] = "and",
) -> dict[str, int]:
    """Retrieves service order ids from the database.

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

        If None, retrieves information for all available.

    descriptions : list[str] | None, optional
        List of descriptions (regex) to retrieve information for.

        If None, retrieves information for all available.

    sap_ids : list[int] | None, optional
        List of SAP IDs to retrieve information for.

        If None, retrieves information for all available.

    statuses : list[str] | None, optional
        List of statuses to retrieve information for.

        If None, retrieves information for all available.

    filter_type : Literal["and", "or"], optional
        How to treat multiple filters. Can be one of ["and", "or"]. By default "and".

    Returns
    -------
    dict[str, int]
        A dictionary in the format {name: id, ...}
    """
    where_query = self._check_get_args(
        names=names,
        descriptions=descriptions,
        sap_ids=sap_ids,
        statuses=statuses,
        filter_type=filter_type,
    )

    query = sql.SQL(
        "SELECT name, id FROM performance.v_service_orders {where_query} ORDER BY name",
    ).format(
        where_query=where_query,
    )

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

    return dict(zip(df["name"].to_list(), df["id"].to_list(), strict=False))

`insert(name=None, status_name=None, sap_id=None, description=None, data_df=None, on_conflict='ignore')` ¶

Inserts a new service order, and ignore if the order already exists.

Fields left as None will not be inserted/updated.

You can either pass individual values to insert a single service order, or pass a DataFrame to insert multiple service orders at once.

Parameters:

name ¶
(str | None, default: None ) –

Name of the service order to be inserted.
status_name ¶
(str | None, default: None ) –

Status name of the service order to be inserted.
sap_id ¶
(int | None, default: None ) –

SAP ID of the service order to be inserted.
description ¶
(str | None, default: None ) –

Description of the service order to be inserted.
data_df ¶
(DataFrame | None, default: None ) –

Polars DataFrame containing multiple service orders to be inserted.

The needed columns are: - name - status_name - sap_id - description

If this is used all the individual parameters will be ignored.
on_conflict ¶
(Literal['ignore', 'update'], default: 'ignore' ) –
Strategy to handle conflicts when inserting data. 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 name of the service order.

Returns:

int | list[int] | None –

If inserting a single service order, returns the ID of the inserted service order.

If inserting multiple service orders, returns a list of IDs of the inserted service orders.

If no service order was inserted (due to conflicts and on_conflict="ignore"), returns None.

Source code in echo_postgres/service_orders.py

@validate_call
def insert(
    self,
    name: str | None = None,
    status_name: str | None = None,
    sap_id: int | None = None,
    description: str | None = None,
    data_df: pl.DataFrame | None = None,
    on_conflict: Literal["ignore", "update"] = "ignore",
) -> int | list[int] | None:
    """Inserts a new service order, and ignore if the order already exists.

    Fields left as None will not be inserted/updated.

    You can either pass individual values to insert a single service order, or pass a DataFrame to insert multiple service orders at once.

    Parameters
    ----------
    name : str | None, optional
        Name of the service order to be inserted.
    status_name : str | None, optional
        Status name of the service order to be inserted.
    sap_id : int | None, optional
        SAP ID of the service order to be inserted.
    description : str | None, optional
        Description of the service order to be inserted.
    data_df : pl.DataFrame | None, optional
        Polars DataFrame containing multiple service orders to be inserted.

        The needed columns are:
        - name
        - status_name
        - sap_id
        - description

        If this is used all the individual parameters will be ignored.
    on_conflict : Literal["ignore", "update"], optional
        Strategy to handle conflicts when inserting data. 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 name of the service order.

    Returns
    -------
    int | list[int] | None
        If inserting a single service order, returns the ID of the inserted service order.

        If inserting multiple service orders, returns a list of IDs of the inserted service orders.

        If no service order was inserted (due to conflicts and on_conflict="ignore"), returns None.

    """
    if data_df is None:
        single_insert = True
        # creating a DataFrame from the individual parameters
        data_df = pl.DataFrame(
            {
                "name": [name],
                "status_name": [status_name],
                "sap_id": [sap_id],
                "description": [description],
            },
            schema={
                "name": pl.Utf8,
                "status_name": pl.Utf8,
                "sap_id": pl.Int64,
                "description": pl.Utf8,
            },
        )
    else:
        single_insert = False
        # check if all needed columns are present
        needed_columns = {"name", "status_name", "sap_id", "description"}
        missing_columns = needed_columns - set(data_df.columns)
        wrong_columns = set(data_df.columns) - needed_columns
        if missing_columns or wrong_columns:
            raise ValueError(
                f"DataFrame must contain the following columns: {needed_columns}. "
                f"Missing columns: {missing_columns}. Wrong columns: {wrong_columns}.",
            )

    # get the status id from the status name
    status_dict = self.status.get_ids(names=data_df["status_name"].unique().to_list())
    if wrong_statuses := set(data_df["status_name"].to_list()) - set(status_dict.keys()):
        raise ValueError(f"Status names not found in the database: {wrong_statuses}")

    # replacing status_name with status_id
    data_df = data_df.with_columns(
        pl.col("status_name")
        .replace_strict(
            status_dict,
            return_dtype=pl.Int64,
        )
        .alias("status_id"),
    )
    data_df = data_df.drop(["status_name"])

    ids_df = self._perfdb.conn.polars_to_sql(
        df=data_df,
        schema="performance",
        table_name="service_orders",
        if_exists="append" if on_conflict == "ignore" else "update",
        return_cols=["id"],
        conflict_cols=["name"],
        ignore_null_cols=single_insert,
    )

    ids = ids_df["id"].to_list()

    logger.debug(
        f"Inserted Service Orders with IDs: {ids}",
    )

    return ids if not single_insert else ids[0] if ids else None

Service Order¶

`ServiceOrders(perfdb)` ¶

`perfdb` ¶

`delete(names, descriptions=None, sap_ids=None, statuses=None, filter_type='and')` ¶

`names` ¶

`descriptions` ¶

`sap_ids` ¶

`statuses` ¶

`filter_type` ¶

`get(names=None, descriptions=None, sap_ids=None, statuses=None, filter_type='and', output_type='dict')` ¶

`names` ¶

`descriptions` ¶

`sap_ids` ¶

`statuses` ¶

`filter_type` ¶

`output_type` ¶

`get_ids(names=None, descriptions=None, sap_ids=None, statuses=None, filter_type='and')` ¶

`names` ¶

`descriptions` ¶

`sap_ids` ¶

`statuses` ¶

`filter_type` ¶

`insert(name=None, status_name=None, sap_id=None, description=None, data_df=None, on_conflict='ignore')` ¶

`name` ¶

`status_name` ¶

`sap_id` ¶

`description` ¶

`data_df` ¶

`on_conflict` ¶

Service Order¶

ServiceOrders(perfdb) ¶

perfdb ¶

delete(names, descriptions=None, sap_ids=None, statuses=None, filter_type='and') ¶

names ¶

descriptions ¶

sap_ids ¶

statuses ¶

filter_type ¶

get(names=None, descriptions=None, sap_ids=None, statuses=None, filter_type='and', output_type='dict') ¶

names ¶

descriptions ¶

sap_ids ¶

statuses ¶

filter_type ¶

output_type ¶

get_ids(names=None, descriptions=None, sap_ids=None, statuses=None, filter_type='and') ¶

names ¶

descriptions ¶

sap_ids ¶

statuses ¶

filter_type ¶

insert(name=None, status_name=None, sap_id=None, description=None, data_df=None, on_conflict='ignore') ¶

name ¶

status_name ¶

sap_id ¶

description ¶

data_df ¶

on_conflict ¶

`ServiceOrders(perfdb)` ¶

`perfdb` ¶

`delete(names, descriptions=None, sap_ids=None, statuses=None, filter_type='and')` ¶

`names` ¶

`descriptions` ¶

`sap_ids` ¶

`statuses` ¶

`filter_type` ¶

`get(names=None, descriptions=None, sap_ids=None, statuses=None, filter_type='and', output_type='dict')` ¶

`names` ¶

`descriptions` ¶

`sap_ids` ¶

`statuses` ¶

`filter_type` ¶

`output_type` ¶

`get_ids(names=None, descriptions=None, sap_ids=None, statuses=None, filter_type='and')` ¶

`names` ¶

`descriptions` ¶

`sap_ids` ¶

`statuses` ¶

`filter_type` ¶

`insert(name=None, status_name=None, sap_id=None, description=None, data_df=None, on_conflict='ignore')` ¶

`name` ¶

`status_name` ¶

`sap_id` ¶

`description` ¶

`data_df` ¶

`on_conflict` ¶