Inventory Withdrawal Items¶

`InventoryWithdrawalItems(perfdb)` ¶

Class used for handling Inventory Withdrawal Items. Can be accessed via perfdb.inventory.withdrawals.items.

This is a read-only view that aggregates transaction items associated with withdrawals, showing withdrawn, consumed, and returned quantities per material.

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

`get(withdrawal_ids=None, center_names=None, storage_location_names=None, service_order_names=None, material_descriptions=None, material_sap_ids=None, withdrawal_statuses=None, is_fully_accounted=None, period=None, filter_type='and', output_type='pl.DataFrame')` ¶

Gets inventory withdrawal items from the aggregated view.

This is a read-only view that aggregates transaction items associated with withdrawals, showing withdrawn, consumed, and returned quantities per material.

The most useful keys/columns returned are:

withdrawal_id
withdrawal_date
withdrawal_status
service_order_name
service_order_sap_id
material_id
material_description
material_sap_id
base_unit
withdrawn_quantity
consumed_quantity
returned_quantity
is_fully_accounted
storage_location_name
center_name
created_by_id
creator_name

Parameters:

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

List of withdrawal IDs to filter the results. By default None.
center_names ¶
(list[str] | None, default: None ) –

List of center names to filter the results. By default None.
storage_location_names ¶
(list[str] | None, default: None ) –

List of storage location names to filter the results. By default None.
service_order_names ¶
(list[str] | None, default: None ) –

List of service order names to filter the results. By default None.
material_descriptions ¶
(list[str] | None, default: None ) –

List of material descriptions to filter the results. By default None.
material_sap_ids ¶
(list[int] | None, default: None ) –

List of material SAP IDs to filter the results. By default None.
withdrawal_statuses ¶
(list[str] | None, default: None ) –

List of withdrawal statuses to filter the results. Valid values: ABERTA, PARCIALMENTE_APONTADA, APONTADA, CANCELADA. By default None.
is_fully_accounted ¶
(bool | None, default: None ) –

Filter by whether the withdrawal item is fully accounted (consumed + returned = withdrawn). By default None.
period ¶
(DateTimeRange | None, default: None ) –

Date range to filter by withdrawal_date. 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: 'pl.DataFrame' ) –

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

Returns:

dict[int, dict[str, Any]] –

In case output_type is "dict", returns a dictionary in the format {withdrawal_id: {material_id: {attribute: value, ...}, ...}, ...}.
DataFrame –

In case output_type is "DataFrame", returns a pandas DataFrame with MultiIndex = (withdrawal_id, material_id).
DataFrame –

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

Source code in echo_postgres/inventory_withdrawal_items.py

@validate_call
def get(
    self,
    withdrawal_ids: list[int] | None = None,
    center_names: list[str] | None = None,
    storage_location_names: list[str] | None = None,
    service_order_names: list[str] | None = None,
    material_descriptions: list[str] | None = None,
    material_sap_ids: list[int] | None = None,
    withdrawal_statuses: list[str] | None = None,
    is_fully_accounted: bool | None = None,
    period: DateTimeRange | None = None,
    filter_type: Literal["and", "or"] = "and",
    output_type: Literal["dict", "DataFrame", "pl.DataFrame"] = "pl.DataFrame",
) -> dict[int, dict[str, Any]] | pd.DataFrame | pl.DataFrame:
    """Gets inventory withdrawal items from the aggregated view.

    This is a read-only view that aggregates transaction items associated with withdrawals,
    showing withdrawn, consumed, and returned quantities per material.

    The most useful keys/columns returned are:

    - withdrawal_id
    - withdrawal_date
    - withdrawal_status
    - service_order_name
    - service_order_sap_id
    - material_id
    - material_description
    - material_sap_id
    - base_unit
    - withdrawn_quantity
    - consumed_quantity
    - returned_quantity
    - is_fully_accounted
    - storage_location_name
    - center_name
    - created_by_id
    - creator_name

    Parameters
    ----------
    withdrawal_ids : list[int] | None, optional
        List of withdrawal IDs to filter the results. By default None.
    center_names : list[str] | None, optional
        List of center names to filter the results. By default None.
    storage_location_names : list[str] | None, optional
        List of storage location names to filter the results. By default None.
    service_order_names : list[str] | None, optional
        List of service order names to filter the results. By default None.
    material_descriptions : list[str] | None, optional
        List of material descriptions to filter the results. By default None.
    material_sap_ids : list[int] | None, optional
        List of material SAP IDs to filter the results. By default None.
    withdrawal_statuses : list[str] | None, optional
        List of withdrawal statuses to filter the results. Valid values: ABERTA, PARCIALMENTE_APONTADA, APONTADA, CANCELADA.
        By default None.
    is_fully_accounted : bool | None, optional
        Filter by whether the withdrawal item is fully accounted (consumed + returned = withdrawn).
        By default None.
    period : DateTimeRange | None, optional
        Date range to filter by withdrawal_date. 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 "pl.DataFrame".

    Returns
    -------
    dict[int, dict[str, Any]]
        In case output_type is "dict", returns a dictionary in the format
        {withdrawal_id: {material_id: {attribute: value, ...}, ...}, ...}.
    pd.DataFrame
        In case output_type is "DataFrame", returns a pandas DataFrame with MultiIndex = (withdrawal_id, material_id).
    pl.DataFrame
        In case output_type is "pl.DataFrame", returns a Polars DataFrame.
    """
    where = self._check_get_args(
        withdrawal_ids=withdrawal_ids,
        center_names=center_names,
        storage_location_names=storage_location_names,
        service_order_names=service_order_names,
        material_descriptions=material_descriptions,
        material_sap_ids=material_sap_ids,
        withdrawal_statuses=withdrawal_statuses,
        is_fully_accounted=is_fully_accounted,
        period=period,
        filter_type=filter_type,
    )

    query = sql.SQL(
        "SELECT * FROM performance.v_inv_withdrawal_items {where} ORDER BY withdrawal_id, material_id",
    ).format(where=where)

    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(["withdrawal_id", "material_id"])

    if output_type == "DataFrame":
        return df

    # nested dict: {withdrawal_id: {material_id: {attrs}}}
    result: dict[int, dict[str, Any]] = {}
    for (wid, mid), row in df.iterrows():
        if wid not in result:
            result[wid] = {}
        result[wid][mid] = row.to_dict()

    return result

Inventory Withdrawal Items¶