Inventory Centers¶
InventoryCenters(perfdb)
¶
Class used for handling Inventory Centers. Can be accessed via perfdb.inventory.centers.
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(name)
¶
Deletes an inventory center from the database.
Parameters:
-
(name¶str) –Name of the inventory center to be deleted.
Returns:
-
int–Number of rows deleted (0 if no inventory center with the given name was found).
Source code in echo_postgres/inventory_centers.py
def delete(self, name: str) -> int:
"""Deletes an inventory center from the database.
Parameters
----------
name : str
Name of the inventory center to be deleted.
Returns
-------
int
Number of rows deleted (0 if no inventory center with the given name was found).
"""
query = sql.SQL("DELETE FROM performance.inv_centers WHERE name = {}").format(sql.Literal(name))
with self._perfdb.conn.reconnect() as conn:
conn.execute(query)
return conn.rowcount
get(names=None, is_active=None, filter_type='and', output_type='pl.DataFrame')
¶
Gets all inventory centers and its attributes. The most useful keys/columns returned are:
- name (index in case of DataFrame output)
- is_active
- city_name
- city_state
Parameters:
-
(names¶list[str] | None, default:None) –List of center names to filter. By default None.
-
(is_active¶bool | None, default:None) –Filter inventory centers by their active status. By default None.
-
(filter_type¶Literal['and', 'or'], default:'and') –How to treat multiple filters. 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, str | int]] | DataFrame | DataFrame–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 with the following format: index = name, columns = [attribute, ...]
Source code in echo_postgres/inventory_centers.py
@validate_call
def get(
self,
names: list[str] | None = None,
is_active: 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 centers and its attributes. The most useful keys/columns returned are:
- name (index in case of DataFrame output)
- is_active
- city_name
- city_state
Parameters
----------
names : list[str] | None, optional
List of center names to filter. By default None.
is_active : bool | None, optional
Filter inventory centers by their active status. By default None.
filter_type : Literal["and", "or"], optional
How to treat multiple filters. 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, str | int]] | pd.DataFrame | pl.DataFrame
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 with the following format: index = name, columns = [attribute, ...]
"""
where = self._check_get_args(names=names, is_active=is_active, filter_type=filter_type)
query = sql.SQL("SELECT * FROM performance.v_inv_centers {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, is_active=None, filter_type='and')
¶
Gets all inventory center IDs indexed by name.
Parameters:
-
(names¶list[str] | None, default:None) –List of center names to filter. By default None.
-
(is_active¶bool | None, default:None) –Filter by active status. By default None.
-
(filter_type¶Literal['and', 'or'], default:'and') –How to treat multiple filters. By default "and".
Returns:
-
dict[str, int]–Dictionary with center names as keys and IDs as values.
Source code in echo_postgres/inventory_centers.py
@validate_call
def get_ids(
self,
names: list[str] | None = None,
is_active: bool | None = None,
filter_type: Literal["and", "or"] = "and",
) -> dict[str, int]:
"""Gets all inventory center IDs indexed by name.
Parameters
----------
names : list[str] | None, optional
List of center names to filter. By default None.
is_active : bool | None, optional
Filter by active status. By default None.
filter_type : Literal["and", "or"], optional
How to treat multiple filters. By default "and".
Returns
-------
dict[str, int]
Dictionary with center names as keys and IDs as values.
"""
where = self._check_get_args(names=names, is_active=is_active, filter_type=filter_type)
query = sql.SQL("SELECT name, id FROM performance.v_inv_centers {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))
insert(name, city_name, is_active=True, on_conflict='ignore')
¶
Inserts a new inventory center into the database.
Parameters:
-
(name¶str) –Name of the inventory center. Must be unique.
-
(city_name¶str) –Name of the city where the inventory center is located. Must exist in the cities table.
-
(is_active¶bool, default:True) –Whether the inventory center is active or not. By default True.
-
(on_conflict¶Literal['ignore', 'update'], default:'ignore') –What to do in case of conflict. Can be one of ["ignore", "update"]. By default "ignore"
Returns:
-
int–ID of the inserted/update inventory center.
-
None–If on_conflict is "ignore" and there is a conflict, returns None.
Source code in echo_postgres/inventory_centers.py
def insert(self, name: str, city_name: str, is_active: bool = True, on_conflict: Literal["ignore", "update"] = "ignore") -> int | None:
"""Inserts a new inventory center into the database.
Parameters
----------
name : str
Name of the inventory center. Must be unique.
city_name : str
Name of the city where the inventory center is located. Must exist in the cities table.
is_active : bool, optional
Whether the inventory center is active or not. By default True.
on_conflict : Literal["ignore", "update"], optional
What to do in case of conflict. Can be one of ["ignore", "update"].
By default "ignore"
Returns
-------
int
ID of the inserted/update inventory center.
None
If on_conflict is "ignore" and there is a conflict, returns None.
"""
# gets the city_id from the city_name
cities = self._perfdb.cities.get(output_type="pl.DataFrame")
city = cities.filter(pl.col("name") == city_name).select("id").to_series()
if city.is_empty():
raise ValueError(f"City with name '{city_name}' does not exist in the cities table.")
city_id = city[0]
# building the query
query = sql.SQL("""
INSERT INTO performance.inv_centers (name, city_id, is_active)
VALUES ({name}, {city_id}, {is_active})
ON CONFLICT (name) DO {on_conflict}
RETURNING id
""").format(
name=sql.Literal(name),
city_id=sql.Literal(city_id),
is_active=sql.Literal(is_active),
on_conflict=sql.SQL("NOTHING")
if on_conflict == "ignore"
else sql.SQL("UPDATE SET city_id = EXCLUDED.city_id, is_active = EXCLUDED.is_active"),
)
with self._perfdb.conn.reconnect() as conn:
cur = conn.cursor()
cur.execute(query)
if cur.rowcount == 0:
return None
return cur.fetchone()[0]