Utilities

Util functions for plexosdb.

class plexosdb.utils.PreparedPropertiesResult(params, collection_properties, metadata_map, normalized_records, deprecated_format_used)

Bases: object

Prepared inputs for bulk property insertion.

Parameters:

  • params (list[tuple[int, int, Any]])

  • collection_properties (list[tuple[str, int]])

  • metadata_map (dict[tuple[int, int, Any], dict[str, Any]])

  • normalized_records (list[dict[str, Any]])

  • deprecated_format_used (bool)

params: list[tuple[int, int, Any]]
collection_properties: list[tuple[str, int]]
metadata_map: dict[tuple[int, int, Any], dict[str, Any]]
normalized_records: list[dict[str, Any]]
deprecated_format_used: bool
plexosdb.utils.batched(iterable, n)

Implement batched iterator.

https://docs.python.org/3/library/itertools.html#itertools.batched

Parameters:

  • iterable (Iterable[Any])

  • n (int)

Return type:

Iterator[tuple[Any, …]]

plexosdb.utils.validate_string(value)

Validate string and convert it to python object.

This function also tries to parse floats or ints.

Parameters:

value (Any) – String value to be converted to Python Object

Return type:

Any

Note

The ast is slow due to the multiple cases. Use it only on simple for loops as this could become a bottleneck.

plexosdb.utils.no_space(a, b)

Collate function for catching strings with spaces.

Parameters:

  • a (str)

  • b (str)

Return type:

int

plexosdb.utils.normalize_names(*args)

Normalize a name or list of names into a unique list of strings.

Parameters:

  • names (str or Iterable[str]) – A string or an iterable of strings to normalize

  • args (str | Iterable[str])

Returns:

A deduplicated list of the provided names

Return type:

list[str]

Raises:

ValueError – If the input is neither a string nor an iterable of strings

plexosdb.utils.get_sql_query(query_name)

Load SQL query from package.

Parameters:

query_name (str) – Name of the query file to load from plexosdb.queries

Returns:

Content of the SQL query file as a string

Return type:

str

plexosdb.utils.prepare_sql_data_params(records, memberships, property_mapping)

Create list of tuples for data ingestion.

Parameters:

  • records (list[dict[str, float]]) – List of records where each record is a dictionary containing ‘name’ and property values

  • memberships (list[dict[str, int]]) – List of membership dictionaries with ‘name’ and ‘membership_id’ keys

  • property_mapping (list[tuple[str, int]]) – List of tuples mapping property names to property IDs

Returns:

List of tuples containing (membership_id, property_id, value) for database insertion

Return type:

list[tuple[int, int, Any]]

plexosdb.utils.create_membership_record(object_ids, child_object_class_id, parent_object_id, parent_object_class_id, collection_id)

Create membership records for database insertion.

Parameters:

  • object_ids (Iterable[int]) – Iterable of child object IDs to create memberships for

  • child_object_class_id (int) – Class ID for the child objects

  • parent_object_id (int) – ID of the parent object

  • parent_object_class_id (int) – Class ID for the parent object

  • collection_id (int) – ID of the collection to which the membership belongs

Returns:

List of dictionaries representing membership records ready for database insertion

Return type:

list[dict[str, int]]

plexosdb.utils.plan_property_inserts(db, records, *, object_class, collection, parent_class)

Prepare SQL parameters for property insertion.

Parameters:
Return type:

PreparedPropertiesResult

plexosdb.utils.insert_property_values(db, params, *, metadata_map=None)

Insert property data and return mapping of data IDs to object names.

Parameters:

  • db (PlexosDB) – Database instance

  • params (list[tuple]) – List of (membership_id, property_id, value) tuples

  • metadata_map (dict | None, optional) – Mapping of params to metadata (band, date_from, date_to), by default None

Returns:

Mapping of (membership_id, property_id, value) to (data_id, obj_name)

Return type:

dict

plexosdb.utils.apply_scenario_tags(db, params, /, *, scenario, chunksize)

Insert scenario tags for property data.

Parameters:

  • db (PlexosDB) – Database instance

  • params (list[tuple]) – List of (membership_id, property_id, value) tuples

  • scenario (str) – Scenario name

  • chunksize (int) – Number of records to process in each batch

Return type:

None

plexosdb.utils.insert_property_texts(db, params, /, *, data_id_map, records, field_name, text_class, metadata_map=None)

Add text data for properties from specified field.

Parameters:

  • db (PlexosDB) – Database instance

  • params (list[tuple]) – List of (membership_id, property_id, value) tuples

  • data_id_map (dict) – Mapping of (membership_id, property_id, value) to (data_id, obj_name)

  • records (list[dict]) – Original records containing the text field

  • field_name (str) – Name of the field in records containing text data

  • text_class (ClassEnum) – ClassEnum for the text data

  • metadata_map (dict | None, optional) – Metadata map keyed by param tuple to drive property-specific text mapping

Return type:

None

plexosdb.utils.build_data_id_map(db, params)

Build mapping of (membership_id, property_id, value) to (data_id, obj_name).

Parameters:

  • db_manager (DBManager) – Database manager instance for executing queries

  • params (list[tuple]) – List of (membership_id, property_id, value) tuples

  • db (SQLiteManager)

Returns:

Mapping of (membership_id, property_id, value) to (data_id, obj_name)

Return type:

dict

plexosdb.utils.get_scenario_id(db, scenario)

Get or create scenario ID.

Parameters:

  • db_manager (DBManager) – Database manager instance

  • scenario (str) – Scenario name

  • db (PlexosDB)

Returns:

Scenario object ID

Return type:

int