Inventory Current Balances

InventoryCurrentBalances(perfdb)

Class used for handling Inventory Current Balances. Can be accessed via perfdb.inventory.current_balances.

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(center_names=None, storage_location_names=None, material_descriptions=None, material_sap_ids=None, is_serviceable_stock=None, filter_type='and', output_type='pl.DataFrame')

Gets inventory current balances.

This is a read-only view that calculates balances from transaction history.

The most useful keys/columns returned are:

• center_name
• storage_location_name
• material_description
• material_sap_id
• base_unit
• total_inflow
• total_outflow
• current_balance
• last_movement_date
• transaction_count

Parameters:

• 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.

• 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.

• is_serviceable_stock

  (bool | None, default: None ) –

  Filter by serviceable stock status. 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 {material_id: {attribute: value, ...}, ...}.

• DataFrame –

  In case output_type is "DataFrame", returns a pandas DataFrame with index = material_id.

• DataFrame –

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

Source code in echo_postgres/inventory_current_balances.py

@validate_call
def get(
    self,
    center_names: list[str] | None = None,
    storage_location_names: list[str] | None = None,
    material_descriptions: list[str] | None = None,
    material_sap_ids: list[int] | None = None,
    is_serviceable_stock: bool | 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 current balances.

    This is a read-only view that calculates balances from transaction history.

    The most useful keys/columns returned are:

    - center_name
    - storage_location_name
    - material_description
    - material_sap_id
    - base_unit
    - total_inflow
    - total_outflow
    - current_balance
    - last_movement_date
    - transaction_count

    Parameters
    ----------
    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.
    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.
    is_serviceable_stock : bool | None, optional
        Filter by serviceable stock status. 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 {material_id: {attribute: value, ...}, ...}.
    pd.DataFrame
        In case output_type is "DataFrame", returns a pandas DataFrame with index = material_id.
    pl.DataFrame
        In case output_type is "pl.DataFrame", returns a Polars DataFrame.
    """
    where = self._check_get_args(
        center_names=center_names,
        storage_location_names=storage_location_names,
        material_descriptions=material_descriptions,
        material_sap_ids=material_sap_ids,
        is_serviceable_stock=is_serviceable_stock,
        filter_type=filter_type,
    )

    query = sql.SQL(
        "SELECT * FROM performance.v_inv_current_balances {where} ORDER BY center_name, storage_location_name, material_description",
    ).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("material_id")

    if output_type == "DataFrame":
        return df

    return df.to_dict(orient="index")