Service Order Notes¶
ServiceOrderNotes(perfdb)
¶
Class used for handling Service Order Note data. Can be accessed via perfdb.service_orders.notes.
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(service_order_names=None, service_order_sap_ids=None, note_descriptions=None, note_sap_ids=None, service_order_statuses=None, note_statuses=None, filter_type='and')
¶
Deletes service order notes from the database.
Be cautious when using this method, as it will permanently delete data.
Parameters:
-
(service_order_names¶list[str] | None, default:None) –List of service order names to retrieve information for.
If None, retrieves information for all available. By default None.
-
(note_descriptions¶list[str] | None, default:None) –List of service order note descriptions to retrieve information for. Will be treated as regex.
If None, retrieves information for all available. By default None.
-
(note_sap_ids¶list[int] | None, default:None) –List of service order note SAP IDs to retrieve information for.
If None, retrieves information for all available. By default None.
-
(service_order_statuses¶list[str] | None, default:None) –List of service order statuses to filter the notes by.
If None, retrieves information for all available. By default None.
-
(note_statuses¶list[str] | None, default:None) –List of note statuses to filter the notes by.
If None, retrieves information for all available. By default None.
-
(filter_type¶Literal['and', 'or'], default:'and') –How to treat multiple filters. Can be one of ["and", "or"]
By default "and"
Source code in echo_postgres/service_order_notes.py
@validate_call
def delete(
self,
service_order_names: list[str] | None = None,
service_order_sap_ids: list[int] | None = None,
note_descriptions: list[str] | None = None,
note_sap_ids: list[int] | None = None,
service_order_statuses: list[str] | None = None,
note_statuses: list[str] | None = None,
filter_type: Literal["and", "or"] = "and",
) -> None:
"""Deletes service order notes from the database.
Be cautious when using this method, as it will permanently delete data.
Parameters
----------
service_order_names : list[str] | None, optional
List of service order names to retrieve information for.
If None, retrieves information for all available. By default None.
note_descriptions : list[str] | None, optional
List of service order note descriptions to retrieve information for. Will be treated as regex.
If None, retrieves information for all available. By default None.
note_sap_ids : list[int] | None, optional
List of service order note SAP IDs to retrieve information for.
If None, retrieves information for all available. By default None.
service_order_statuses : list[str] | None, optional
List of service order statuses to filter the notes by.
If None, retrieves information for all available. By default None.
note_statuses : list[str] | None, optional
List of note statuses to filter the notes by.
If None, retrieves information for all available. By default None.
filter_type : Literal["and", "or"], optional
How to treat multiple filters. Can be one of ["and", "or"]
By default "and"
"""
# checking if at least one filter is provided
if not any(
[
service_order_names,
service_order_sap_ids,
note_descriptions,
note_sap_ids,
service_order_statuses,
note_statuses,
],
):
raise ValueError("At least one filter must be provided to delete service order notes.")
# getting ids to delete
ids = self.get_ids(
service_order_names=service_order_names,
service_order_sap_ids=service_order_sap_ids,
note_descriptions=note_descriptions,
note_sap_ids=note_sap_ids,
service_order_statuses=service_order_statuses,
note_statuses=note_statuses,
filter_type=filter_type,
)
if not ids:
logger.debug("No service order notes found to delete based on the provided filters.")
return
query = sql.SQL(
"DELETE FROM performance.service_order_notes WHERE id = ANY({ids})",
).format(
ids=sql.Literal(list(ids.values())),
)
with self._perfdb.conn.reconnect() as conn:
result = conn.execute(query)
rows = result.rowcount if result else 0
logger.debug(f"Deleted {rows} service order notes from the database.")
get(service_order_names=None, service_order_sap_ids=None, note_descriptions=None, note_sap_ids=None, service_order_statuses=None, note_statuses=None, filter_type='and', output_type='dict')
¶
Retrieves service order note information from the database.
The most useful keys/columns returned are:
- service_order_id
- service_order_name
- service_order_sap_id
- note_id
- note_sap_id (is the index if output_type is "DataFrame")
- note_description
- note_status_id
- note_status_name
- service_order_status_id
- service_order_status_name
Parameters:
-
(service_order_names¶list[str] | None, default:None) –List of service order names to retrieve information for.
If None, retrieves information for all available. By default None.
-
(service_order_sap_ids¶list[int] | None, default:None) –List of service order SAP IDs to retrieve information for.
If None, retrieves information for all available. By default None.
-
(note_descriptions¶list[str] | None, default:None) –List of service order note descriptions to retrieve information for. Will be treated as regex.
If None, retrieves information for all available. By default None.
-
(note_sap_ids¶list[int] | None, default:None) –List of service order note SAP IDs to retrieve information for.
If None, retrieves information for all available. By default None.
-
(service_order_statuses¶list[str] | None, default:None) –List of service order statuses to filter the notes by.
If None, retrieves information for all available. By default None.
-
(note_statuses¶list[str] | None, default:None) –List of note statuses to filter the notes by.
If None, retrieves information for all available. 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:'dict') –Output type of the data. Can be one of ["dict", "DataFrame", "pl.DataFrame"] By default "dict"
Returns:
-
dict[str, dict[str, Any]]–In case output_type is "dict", returns a dictionary in the format {note_sap_id: {attribute: value, ...}, ...}
-
DataFrame–In case output_type is "DataFrame", returns a DataFrame with the following format: index = note_sap_id, columns = [attribute, ...]
-
DataFrame–In case output_type is "pl.DataFrame", returns a Polars DataFrame
Source code in echo_postgres/service_order_notes.py
@validate_call
def get(
self,
service_order_names: list[str] | None = None,
service_order_sap_ids: list[int] | None = None,
note_descriptions: list[str] | None = None,
note_sap_ids: list[int] | None = None,
service_order_statuses: list[str] | None = None,
note_statuses: list[str] | None = None,
filter_type: Literal["and", "or"] = "and",
output_type: Literal["dict", "DataFrame", "pl.DataFrame"] = "dict",
) -> dict[str, dict[str, Any]] | pd.DataFrame | pl.DataFrame:
"""Retrieves service order note information from the database.
The most useful keys/columns returned are:
- service_order_id
- service_order_name
- service_order_sap_id
- note_id
- note_sap_id (is the index if output_type is "DataFrame")
- note_description
- note_status_id
- note_status_name
- service_order_status_id
- service_order_status_name
Parameters
----------
service_order_names : list[str] | None, optional
List of service order names to retrieve information for.
If None, retrieves information for all available. By default None.
service_order_sap_ids : list[int] | None, optional
List of service order SAP IDs to retrieve information for.
If None, retrieves information for all available. By default None.
note_descriptions : list[str] | None, optional
List of service order note descriptions to retrieve information for. Will be treated as regex.
If None, retrieves information for all available. By default None.
note_sap_ids : list[int] | None, optional
List of service order note SAP IDs to retrieve information for.
If None, retrieves information for all available. By default None.
service_order_statuses : list[str] | None, optional
List of service order statuses to filter the notes by.
If None, retrieves information for all available. By default None.
note_statuses : list[str] | None, optional
List of note statuses to filter the notes by.
If None, retrieves information for all available. 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 "dict"
Returns
-------
dict[str, dict[str, Any]]
In case output_type is "dict", returns a dictionary in the format {note_sap_id: {attribute: value, ...}, ...}
pd.DataFrame
In case output_type is "DataFrame", returns a DataFrame with the following format: index = note_sap_id, columns = [attribute, ...]
pl.DataFrame
In case output_type is "pl.DataFrame", returns a Polars DataFrame
"""
where_query = self._check_get_args(
service_order_names=service_order_names,
service_order_sap_ids=service_order_sap_ids,
note_descriptions=note_descriptions,
note_sap_ids=note_sap_ids,
service_order_statuses=service_order_statuses,
note_statuses=note_statuses,
filter_type=filter_type,
)
query = sql.SQL(
"SELECT * FROM performance.v_service_order_notes {where_query} ORDER BY service_order_name, note_sap_id",
).format(
where_query=where_query,
)
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("note_sap_id")
if output_type == "DataFrame":
return df
return df.to_dict(orient="index")
get_ids(service_order_names=None, service_order_sap_ids=None, note_descriptions=None, note_sap_ids=None, service_order_statuses=None, note_statuses=None, filter_type='and')
¶
Retrieves service order note ids from the database.
Parameters:
-
(service_order_names¶list[str] | None, default:None) –List of service order names to retrieve information for.
If None, retrieves information for all available. By default None.
-
(note_descriptions¶list[str] | None, default:None) –List of service order note descriptions to retrieve information for. Will be treated as regex.
If None, retrieves information for all available. By default None.
-
(note_sap_ids¶list[int] | None, default:None) –List of service order note SAP IDs to retrieve information for.
If None, retrieves information for all available. By default None.
-
(service_order_statuses¶list[str] | None, default:None) –List of service order statuses to filter the notes by.
If None, retrieves information for all available. By default None.
-
(note_statuses¶list[str] | None, default:None) –List of note statuses to filter the notes by.
If None, retrieves information for all available. 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 in the format {note_sap_id: note_id, ...}
Source code in echo_postgres/service_order_notes.py
@validate_call
def get_ids(
self,
service_order_names: list[str] | None = None,
service_order_sap_ids: list[int] | None = None,
note_descriptions: list[str] | None = None,
note_sap_ids: list[int] | None = None,
service_order_statuses: list[str] | None = None,
note_statuses: list[str] | None = None,
filter_type: Literal["and", "or"] = "and",
) -> dict[str, int]:
"""Retrieves service order note ids from the database.
Parameters
----------
service_order_names : list[str] | None, optional
List of service order names to retrieve information for.
If None, retrieves information for all available. By default None.
note_descriptions : list[str] | None, optional
List of service order note descriptions to retrieve information for. Will be treated as regex.
If None, retrieves information for all available. By default None.
note_sap_ids : list[int] | None, optional
List of service order note SAP IDs to retrieve information for.
If None, retrieves information for all available. By default None.
service_order_statuses : list[str] | None, optional
List of service order statuses to filter the notes by.
If None, retrieves information for all available. By default None.
note_statuses : list[str] | None, optional
List of note statuses to filter the notes by.
If None, retrieves information for all available. 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 in the format {note_sap_id: note_id, ...}
"""
where = self._check_get_args(
service_order_names=service_order_names,
service_order_sap_ids=service_order_sap_ids,
note_descriptions=note_descriptions,
note_sap_ids=note_sap_ids,
service_order_statuses=service_order_statuses,
note_statuses=note_statuses,
filter_type=filter_type,
)
query = sql.SQL(
"SELECT note_sap_id, note_id FROM performance.v_service_order_notes {where_query} ORDER BY note_sap_id",
).format(
where_query=where,
)
with self._perfdb.conn.reconnect() as conn:
df = conn.read_to_polars(query)
return dict(zip(df["note_sap_id"], df["note_id"], strict=False))
insert(sap_id=None, status_name=None, description=None, service_order_name=None, data_df=None, on_conflict='ignore')
¶
Inserts a new service order note, linked with an existing service order id.
You can either pass individual values to insert a single service order note, or pass a DataFrame to insert multiple service order notes at once.
Parameters:
-
(sap_id¶int | None, default:None) –SAP ID of the service order note to be inserted.
-
(status_name¶str | None, default:None) –Status name of the service order note to be inserted. Must already exist in the database.
-
(description¶str | None, default:None) –Description of the service order note to be inserted.
-
(service_order_name¶str | None, default:None) –Name of the service order to link the note to. Must already exist in the database.
-
(data_df¶DataFrame | None, default:None) –Polars DataFrame containing multiple service order notes to be inserted.
The needed columns are: - sap_id - status_name - description (optional) - service_order_name (optional)
If this is used all the individual parameters will be ignored.
-
(on_conflict¶Literal['ignore', 'update'], default:'ignore') –Action to take in case of conflict. Can be one of:
- "ignore": ignores the new data if a conflict occurs (default)
- "update": updates the existing data with the new data in case of conflict
The conflict will be determined based on the sap_id of the service order note.
Returns:
-
int | list[int] | None–If inserting a single service order note, returns the ID of the inserted service order note.
If inserting multiple service order notes, returns a list of IDs of the inserted service order notes.
If no service order note was inserted (due to conflicts and on_conflict="ignore"), returns None.
Source code in echo_postgres/service_order_notes.py
@validate_call
def insert(
self,
sap_id: int | None = None,
status_name: str | None = None,
description: str | None = None,
service_order_name: str | None = None,
data_df: pl.DataFrame | None = None,
on_conflict: Literal["ignore", "update"] = "ignore",
) -> int | list[int] | None:
"""Inserts a new service order note, linked with an existing service order id.
You can either pass individual values to insert a single service order note, or pass a DataFrame to insert multiple service order notes at once.
Parameters
----------
sap_id : int | None, optional
SAP ID of the service order note to be inserted.
status_name : str | None, optional
Status name of the service order note to be inserted. Must already exist in the database.
description : str | None, optional
Description of the service order note to be inserted.
service_order_name : str | None, optional
Name of the service order to link the note to. Must already exist in the database.
data_df : pl.DataFrame | None, optional
Polars DataFrame containing multiple service order notes to be inserted.
The needed columns are:
- sap_id
- status_name
- description (optional)
- service_order_name (optional)
If this is used all the individual parameters will be ignored.
on_conflict : Literal["ignore", "update"], optional
Action to take in case of conflict. Can be one of:
- "ignore": ignores the new data if a conflict occurs (default)
- "update": updates the existing data with the new data in case of conflict
The conflict will be determined based on the sap_id of the service order note.
Returns
-------
int | list[int] | None
If inserting a single service order note, returns the ID of the inserted service order note.
If inserting multiple service order notes, returns a list of IDs of the inserted service order notes.
If no service order note was inserted (due to conflicts and on_conflict="ignore"), returns None.
"""
if data_df is None:
single_insert = True
# creating a DataFrame from the individual parameters
data_df = pl.DataFrame(
{
"sap_id": [sap_id],
"status_name": [status_name],
"description": [description],
"service_order_name": [service_order_name],
},
schema={
"sap_id": pl.Int64,
"status_name": pl.Utf8,
"description": pl.Utf8,
"service_order_name": pl.Utf8,
},
)
else:
single_insert = False
# check if all needed columns are present
required_columns = {"sap_id", "status_name"}
optional_columns = {"description", "service_order_name"}
all_allowed_columns = required_columns | optional_columns
missing_columns = required_columns - set(data_df.columns)
wrong_columns = set(data_df.columns) - all_allowed_columns
if missing_columns or wrong_columns:
raise ValueError(
f"DataFrame must contain the following columns: {required_columns}, "
f"and optionally: {optional_columns}. "
f"Missing columns: {missing_columns}. Wrong columns: {wrong_columns}.",
)
# add missing optional columns
if "description" not in data_df.columns:
data_df = data_df.with_columns(pl.lit(None, dtype=pl.Utf8).alias("description"))
if "service_order_name" not in data_df.columns:
data_df = data_df.with_columns(pl.lit(None, dtype=pl.Utf8).alias("service_order_name"))
# get the status id from the status name
status_dict = self._perfdb.service_orders.status.get_ids(names=data_df["status_name"].unique().to_list())
if wrong_statuses := set(data_df["status_name"].to_list()) - set(status_dict.keys()):
raise ValueError(f"Status names not found in the database: {wrong_statuses}")
# replacing status_name with status_id
data_df = data_df.with_columns(
pl.col("status_name")
.replace_strict(
status_dict,
return_dtype=pl.Int64,
)
.alias("status_id"),
)
data_df = data_df.drop(["status_name"])
# getting service order ids
unique_service_order_names = data_df["service_order_name"].unique().drop_nulls().to_list()
if unique_service_order_names:
service_order_ids = self._perfdb.service_orders.get_ids(
names=unique_service_order_names,
)
if not service_order_ids:
raise ValueError(f"Service orders with names {unique_service_order_names} do not exist in the database.")
if wrong_service_orders := set(unique_service_order_names) - set(service_order_ids.keys()):
raise ValueError(f"Service order names not found in the database: {wrong_service_orders}")
# replacing service_order_name with service_order_id
data_df = data_df.with_columns(
pl.col("service_order_name")
.replace_strict(
service_order_ids,
return_dtype=pl.Int64,
)
.alias("service_order_id"),
)
else:
data_df = data_df.with_columns(pl.lit(None, dtype=pl.Int64).alias("service_order_id"))
data_df = data_df.drop(["service_order_name"])
ids_df = self._perfdb.conn.polars_to_sql(
df=data_df,
schema="performance",
table_name="service_order_notes",
if_exists="append" if on_conflict == "ignore" else "update",
return_cols=["id"],
conflict_cols=["sap_id"],
ignore_null_cols=single_insert,
)
ids = ids_df["id"].to_list()
logger.debug(
f"Inserted Service Order Notes with IDs: {ids}",
)
return ids if not single_insert else ids[0] if ids else None
update(note_id=None, description=None, service_order_name=None, sap_id=None, status_name=None, data_df=None)
¶
Updates an existing service order note in the database.
You can either pass individual values to update a single service order note, or pass a DataFrame to update multiple service order notes at once.
Parameters:
-
(note_id¶int | None, default:None) –ID of the service order note to be updated.
-
(description¶str | None, default:None) –New description for the service order note. If None, the description will not be updated.
-
(service_order_name¶str | None, default:None) –New service order name to link the note to. If None, the service order will not be updated.
-
(sap_id¶int | None, default:None) –New SAP ID for the service order note. If None, the SAP ID will not be updated.
-
(status_name¶str | None, default:None) –New status name for the service order note. If None, the status will not be updated.
-
(data_df¶DataFrame | None, default:None) –Polars DataFrame containing multiple service order notes to be updated.
The required column is: - note_id
The optional columns are: - sap_id - status_name - description - service_order_name
If this is used all the individual parameters will be ignored.
Returns:
-
list[int] | int | None–In case a DataFrame was provided, returns a list of IDs of the updated service order notes.
In case individual parameters were provided, returns the ID of the updated service order note or None if no update was made.
Source code in echo_postgres/service_order_notes.py
@validate_call
def update(
self,
note_id: int | None = None,
description: str | None = None,
service_order_name: str | None = None,
sap_id: int | None = None,
status_name: str | None = None,
data_df: pl.DataFrame | None = None,
) -> list[int] | int | None:
"""Updates an existing service order note in the database.
You can either pass individual values to update a single service order note, or pass a DataFrame to update multiple service order notes at once.
Parameters
----------
note_id : int | None, optional
ID of the service order note to be updated.
description : str | None, optional
New description for the service order note. If None, the description will not be updated.
service_order_name : str | None, optional
New service order name to link the note to. If None, the service order will not be updated.
sap_id : int | None, optional
New SAP ID for the service order note. If None, the SAP ID will not be updated.
status_name : str | None, optional
New status name for the service order note. If None, the status will not be updated.
data_df : pl.DataFrame | None, optional
Polars DataFrame containing multiple service order notes to be updated.
The required column is:
- note_id
The optional columns are:
- sap_id
- status_name
- description
- service_order_name
If this is used all the individual parameters will be ignored.
Returns
-------
list[int] | int | None
In case a DataFrame was provided, returns a list of IDs of the updated service order notes.
In case individual parameters were provided, returns the ID of the updated service order note or None if no update was made.
"""
if data_df is None:
single_update = True
# check if at least one field to update is provided
if not any([description, service_order_name, sap_id, status_name]):
raise ValueError("At least one field to update must be provided.")
# creating a DataFrame from the individual parameters
data_df = pl.DataFrame(
{
"id": [note_id],
"sap_id": [sap_id],
"status_name": [status_name],
"description": [description],
"service_order_name": [service_order_name],
},
schema={
"id": pl.Int64,
"sap_id": pl.Int64,
"status_name": pl.Utf8,
"description": pl.Utf8,
"service_order_name": pl.Utf8,
},
)
else:
single_update = False
# check if all needed columns are present
required_columns = {"id"}
optional_columns = {"sap_id", "status_name", "description", "service_order_name"}
all_allowed_columns = required_columns | optional_columns
missing_columns = required_columns - set(data_df.columns)
wrong_columns = set(data_df.columns) - all_allowed_columns
if missing_columns or wrong_columns:
raise ValueError(
f"DataFrame must contain the following columns: {required_columns}, "
f"and optionally: {optional_columns}. "
f"Missing columns: {missing_columns}. Wrong columns: {wrong_columns}.",
)
# add missing optional columns
if "sap_id" not in data_df.columns:
data_df = data_df.with_columns(pl.lit(None, dtype=pl.Int64).alias("sap_id"))
if "description" not in data_df.columns:
data_df = data_df.with_columns(pl.lit(None, dtype=pl.Utf8).alias("description"))
if "status_name" not in data_df.columns:
data_df = data_df.with_columns(pl.lit(None, dtype=pl.Utf8).alias("status_name"))
if "service_order_name" not in data_df.columns:
data_df = data_df.with_columns(pl.lit(None, dtype=pl.Utf8).alias("service_order_name"))
# get the status id from the status name (only for non-null values)
unique_status_names = data_df["status_name"].unique().drop_nulls().to_list()
if unique_status_names:
status_dict = self._perfdb.service_orders.status.get_ids(names=unique_status_names)
if wrong_statuses := set(unique_status_names) - set(status_dict.keys()):
raise ValueError(f"Status names not found in the database: {wrong_statuses}")
# replacing status_name with status_id
data_df = data_df.with_columns(
pl.col("status_name")
.replace_strict(
status_dict,
return_dtype=pl.Int64,
)
.alias("status_id"),
)
else:
data_df = data_df.with_columns(pl.lit(None, dtype=pl.Int64).alias("status_id"))
data_df = data_df.drop(["status_name"])
# getting service order ids (only for non-null values)
unique_service_order_names = data_df["service_order_name"].unique().drop_nulls().to_list()
if unique_service_order_names:
service_order_ids = self._perfdb.service_orders.get_ids(
names=unique_service_order_names,
)
if not service_order_ids:
raise ValueError(f"Service orders with names {unique_service_order_names} do not exist in the database.")
if wrong_service_orders := set(unique_service_order_names) - set(service_order_ids.keys()):
raise ValueError(f"Service order names not found in the database: {wrong_service_orders}")
# replacing service_order_name with service_order_id
data_df = data_df.with_columns(
pl.col("service_order_name")
.replace_strict(
service_order_ids,
return_dtype=pl.Int64,
)
.alias("service_order_id"),
)
else:
data_df = data_df.with_columns(pl.lit(None, dtype=pl.Int64).alias("service_order_id"))
data_df = data_df.drop(["service_order_name"])
# Remove rows where all update fields are null (nothing to update)
data_df = data_df.filter(
pl.any_horizontal(
pl.col("sap_id").is_not_null(),
pl.col("status_id").is_not_null(),
pl.col("description").is_not_null(),
pl.col("service_order_id").is_not_null(),
),
)
if data_df.is_empty():
logger.debug("No service order notes to update (all update fields are null).")
return None
ids_df = self._perfdb.conn.polars_to_sql(
df=data_df,
schema="performance",
table_name="service_order_notes",
if_exists="update_only",
return_cols=["id"],
conflict_cols=["id"],
ignore_null_cols=single_update,
)
ids = ids_df["id"].to_list()
logger.debug(
f"Updated Service Order Notes with IDs: {ids}",
)
return ids if not single_update else ids[0] if ids else None