Skip to content

Inventory Transaction Types

InventoryTransactionTypes(perfdb)

Class used for handling Inventory Transaction Types. Can be accessed via perfdb.inventory.transactions.types.

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(names=None, directions=None, is_regular_transaction=None, is_adjustment=None, is_inventory_diff=None, filter_type='and', output_type='pl.DataFrame')

Gets all inventory transaction types and their attributes.

The most useful keys/columns returned are:

  • id
  • name
  • direction
  • is_regular_transaction
  • is_adjustment
  • is_inventory_diff
  • description

Parameters:

  • names

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

    List of transaction type names to filter the results. By default None.

  • directions

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

    List of directions to filter the results. By default None.

  • is_regular_transaction

    (bool | None, default: None ) –

    Filter by regular transaction flag. By default None.

  • is_adjustment

    (bool | None, default: None ) –

    Filter by adjustment flag. By default None.

  • is_inventory_diff

    (bool | None, default: None ) –

    Filter by inventory difference flag. 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[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 pandas DataFrame with index = name.

  • DataFrame –

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

Source code in echo_postgres/inventory_transaction_types.py 
@validate_call
def get(
    self,
    names: list[str] | None = None,
    directions: list[str] | None = None,
    is_regular_transaction: bool | None = None,
    is_adjustment: bool | None = None,
    is_inventory_diff: bool | None = None,
    filter_type: Literal["and", "or"] = "and",
    output_type: Literal["dict", "DataFrame", "pl.DataFrame"] = "pl.DataFrame",
) -> dict[str, dict[str, Any]] | pd.DataFrame | pl.DataFrame:
    """Gets all inventory transaction types and their attributes.

    The most useful keys/columns returned are:

    - id
    - name
    - direction
    - is_regular_transaction
    - is_adjustment
    - is_inventory_diff
    - description

    Parameters
    ----------
    names : list[str] | None, optional
        List of transaction type names to filter the results. By default None.
    directions : list[str] | None, optional
        List of directions to filter the results. By default None.
    is_regular_transaction : bool | None, optional
        Filter by regular transaction flag. By default None.
    is_adjustment : bool | None, optional
        Filter by adjustment flag. By default None.
    is_inventory_diff : bool | None, optional
        Filter by inventory difference flag. 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[str, dict[str, Any]]
        In case output_type is "dict", returns a dictionary in the format {name: {attribute: value, ...}, ...}.
    pd.DataFrame
        In case output_type is "DataFrame", returns a pandas DataFrame with index = name.
    pl.DataFrame
        In case output_type is "pl.DataFrame", returns a Polars DataFrame.
    """
    where = self._check_get_args(
        names=names,
        directions=directions,
        is_regular_transaction=is_regular_transaction,
        is_adjustment=is_adjustment,
        is_inventory_diff=is_inventory_diff,
        filter_type=filter_type,
    )

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

    if output_type == "DataFrame":
        return df

    return df.to_dict(orient="index")

get_ids(names=None, directions=None, is_regular_transaction=None, is_adjustment=None, is_inventory_diff=None, filter_type='and')

Gets all inventory transaction types and their respective ids.

Parameters:

  • names

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

    List of transaction type names to filter the results. By default None.

  • directions

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

    List of directions to filter the results. By default None.

  • is_regular_transaction

    (bool | None, default: None ) –

    Filter by regular transaction flag. By default None.

  • is_adjustment

    (bool | None, default: None ) –

    Filter by adjustment flag. By default None.

  • is_inventory_diff

    (bool | None, default: None ) –

    Filter by inventory difference flag. 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 with all transaction types and their respective ids in the format {name: id, ...}.

Source code in echo_postgres/inventory_transaction_types.py 
@validate_call
def get_ids(
    self,
    names: list[str] | None = None,
    directions: list[str] | None = None,
    is_regular_transaction: bool | None = None,
    is_adjustment: bool | None = None,
    is_inventory_diff: bool | None = None,
    filter_type: Literal["and", "or"] = "and",
) -> dict[str, int]:
    """Gets all inventory transaction types and their respective ids.

    Parameters
    ----------
    names : list[str] | None, optional
        List of transaction type names to filter the results. By default None.
    directions : list[str] | None, optional
        List of directions to filter the results. By default None.
    is_regular_transaction : bool | None, optional
        Filter by regular transaction flag. By default None.
    is_adjustment : bool | None, optional
        Filter by adjustment flag. By default None.
    is_inventory_diff : bool | None, optional
        Filter by inventory difference flag. 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 with all transaction types and their respective ids in the format {name: id, ...}.
    """
    where = self._check_get_args(
        names=names,
        directions=directions,
        is_regular_transaction=is_regular_transaction,
        is_adjustment=is_adjustment,
        is_inventory_diff=is_inventory_diff,
        filter_type=filter_type,
    )

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

    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))