Inventory Withdrawal Service Orders¶
InventoryWithdrawalServiceOrders(perfdb)
¶
Class used for handling Inventory Withdrawal Service Orders. Can be accessed via perfdb.inventory.withdrawals.service_orders.
Parameters:
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(mapping_dict)
¶
Deletes specific service orders from specific withdrawals in bulk.
This method acts as a wrapper for the fn_inv_delete_withdrawals_service_orders
database function. It is idempotent; if the relationship does not exist,
it safely ignores it.
Parameters:
-
(mapping_dict¶dict[int, list[str]]) –A dictionary mapping withdrawal IDs to a list of service order names to unlink. Format: {withdrawal_id: ["service_order_name_1", "service_order_name_2", ...]} Example: {50: ["OS-123", "OS-456"]}
Returns:
-
None–
Raises:
-
ValueError–If the database function returns false (e.g., if a service order name is misspelled and cannot be looked up).
Source code in echo_postgres/inventory_withdrawal_service_orders.py
@validate_call
def delete(
self,
mapping_dict: dict[int, list[str]],
) -> None:
"""Deletes specific service orders from specific withdrawals in bulk.
This method acts as a wrapper for the `fn_inv_delete_withdrawals_service_orders`
database function. It is idempotent; if the relationship does not exist,
it safely ignores it.
Parameters
----------
mapping_dict : dict[int, list[str]]
A dictionary mapping withdrawal IDs to a list of service order names to unlink.
Format: {withdrawal_id: ["service_order_name_1", "service_order_name_2", ...]}
Example: {50: ["OS-123", "OS-456"]}
Returns
-------
None
Raises
------
ValueError
If the database function returns false (e.g., if a service order name
is misspelled and cannot be looked up).
"""
# Serialize dict to JSON bytes, then decode to string
json_payload = orjson.dumps(mapping_dict, option=orjson.OPT_NON_STR_KEYS).decode("utf-8")
# Notice the ::jsonb cast in the SQL string
query = sql.SQL(
"SELECT performance.fn_inv_delete_withdrawals_service_orders({payload}::jsonb)",
).format(
payload=sql.Literal(json_payload),
)
result = self._perfdb.conn.read_to_polars(query)
# In case of false result, it means the delete failed (probably due to invalid lookups)
if not result["fn_inv_delete_withdrawals_service_orders"][0]:
raise ValueError(
"Failed to delete withdrawal service order relationships. Please check the provided service order names.",
f"Database messages: {self._perfdb.conn.notices}",
)
logger.debug(f"Successfully unmapped service orders for {len(mapping_dict)} withdrawal(s).")
get(withdrawal_ids=None, service_order_names=None, service_order_ids=None, filter_type='and', output_type='pl.DataFrame')
¶
Gets service orders associated with inventory withdrawals.
The most useful keys/columns returned are:
- withdrawal_id
- service_order_name
- service_order_id
Parameters:
-
(withdrawal_ids¶list[int] | None, default:None) –List of withdrawal IDs to filter the query. By default None.
-
(service_order_names¶list[str] | None, default:None) –List of service order names to filter the query. By default None.
-
(service_order_ids¶list[int] | None, default:None) –List of service order IDs to filter the query. 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:
-
DataFrame–In case output_type is "DataFrame", returns a pandas DataFrame.
-
DataFrame–In case output_type is "pl.DataFrame", returns a Polars DataFrame.
Source code in echo_postgres/inventory_withdrawal_service_orders.py
@validate_call
def get(
self,
withdrawal_ids: list[int] | None = None,
service_order_names: list[str] | None = None,
service_order_ids: list[int] | None = None,
filter_type: Literal["and", "or"] = "and",
output_type: Literal["DataFrame", "pl.DataFrame"] = "pl.DataFrame",
) -> pd.DataFrame | pl.DataFrame:
"""Gets service orders associated with inventory withdrawals.
The most useful keys/columns returned are:
- withdrawal_id
- service_order_name
- service_order_id
Parameters
----------
withdrawal_ids : list[int] | None
List of withdrawal IDs to filter the query. By default None.
service_order_names : list[str] | None
List of service order names to filter the query. By default None.
service_order_ids : list[int] | None
List of service order IDs to filter the query. 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
-------
pd.DataFrame
In case output_type is "DataFrame", returns a pandas DataFrame.
pl.DataFrame
In case output_type is "pl.DataFrame", returns a Polars DataFrame.
"""
where = self._check_get_args(
withdrawal_ids=withdrawal_ids,
service_order_names=service_order_names,
service_order_ids=service_order_ids,
filter_type=filter_type,
)
query = sql.SQL(
"SELECT * FROM performance.v_inv_withdrawal_service_orders {where} ORDER BY withdrawal_id, service_order_name",
).format(where=where)
df = self._perfdb.conn.read_to_polars(query, schema_overrides=self._cols_schema)
if output_type == "pl.DataFrame":
return df
df = df.to_pandas(use_pyarrow_extension_array=True)
return df
insert(mapping_dict)
¶
Inserts multiple service orders to multiple withdrawals in bulk.
This method acts as a wrapper for the fn_inv_insert_withdrawals_service_orders
database function, converting the Python dictionary into a JSON string using orjson.
Parameters:
-
(mapping_dict¶dict[int, list[str]]) –A dictionary mapping withdrawal IDs to a list of service order names. Format: {withdrawal_id: ["service_order_name_1", "service_order_name_2", ...]} Example: {50: ["OS-123", "OS-456"], 51: ["OS-999"]}
Returns:
-
None–
Raises:
-
ValueError–If the database function returns false (e.g., if a withdrawal ID or service order name does not exist).
Source code in echo_postgres/inventory_withdrawal_service_orders.py
@validate_call
def insert(
self,
mapping_dict: dict[int, list[str]],
) -> None:
"""Inserts multiple service orders to multiple withdrawals in bulk.
This method acts as a wrapper for the `fn_inv_insert_withdrawals_service_orders`
database function, converting the Python dictionary into a JSON string using orjson.
Parameters
----------
mapping_dict : dict[int, list[str]]
A dictionary mapping withdrawal IDs to a list of service order names.
Format: {withdrawal_id: ["service_order_name_1", "service_order_name_2", ...]}
Example: {50: ["OS-123", "OS-456"], 51: ["OS-999"]}
Returns
-------
None
Raises
------
ValueError
If the database function returns false (e.g., if a withdrawal ID or
service order name does not exist).
"""
# Serialize dict to JSON bytes, then decode to string
json_payload = orjson.dumps(mapping_dict, option=orjson.OPT_NON_STR_KEYS).decode("utf-8")
# Notice the ::jsonb cast in the SQL string to safely convert the text payload
query = sql.SQL(
"SELECT performance.fn_inv_insert_withdrawals_service_orders({payload}::jsonb)",
).format(
payload=sql.Literal(json_payload),
)
result = self._perfdb.conn.read_to_polars(query)
# In case of false result, it means the insert failed (probably due to invalid lookups)
if not result["fn_inv_insert_withdrawals_service_orders"][0]:
# Get messages/warnings from the database connection for debugging
raise ValueError(
"Failed to insert withdrawal service order relationships. "
"Please check the provided withdrawal IDs and service order names.",
f"Database messages: {self._perfdb.conn.notices}",
)
logger.debug(f"Successfully mapped service orders for {len(mapping_dict)} withdrawal(s).")