crud

CRUD operations on DB model

Functions:

• execute_text_sql_stmt –

  Function to execute text SQL statement and return result

• get_all_devices –

  Function to get all registered devices

• get_last_record –

  Function to get last record from DB for a given device_id or device_name

• get_latest_locations –

  Function to get latest locations stored in DB for a given

• get_locations –

  Function to select and return all locations stored in DB

• insert –

  Function to insert a new record into DB

execute_text_sql_stmt

execute_text_sql_stmt(sql_stmt: str | None = None, mapping: bool = True) -> Any | None

Function to execute text SQL statement and return result

Parameters:

• sql_stmt

  (str | None, default: None ) –

  Input SQL statement

• mapping

  (bool, default: True ) –

  True if results must have columns, False otherwise (default := True)

Returns:

• Any | None –

  Result of SQL statement

get_all_devices

get_all_devices() -> List[Row | RowMapping]

Function to get all registered devices

Returns:

• List[Row] –

  List of all registered devices

get_last_record

get_last_record(session: Session, cond_name: Literal['device_id', 'device_name'] = 'device_name', cond_value: str | Any | None = None) -> DeviceLocation | None

Function to get last record from DB for a given device_id or device_name (Returning only .geometry attribute, not the entire record)

Parameters:

• session

  (Session) –

  SQLAlchemy session (Session object)

• cond_name

  (Literal['device_id', 'device_name'], default: 'device_name' ) –

  Column name of conditional filtering

• cond_value

  (str | Any | None, default: None ) –

  Conditional filtering value

Returns:

• Any | None –

  Last record from DB matching condition, None otherwise

get_latest_locations

get_latest_locations(device_name: str | None = None) -> List[Dict[str, Any]]

Function to get latest locations stored in DB for a given

Parameters:

• device_name

  (str | None, default: None ) –

  Device name

Returns:

• List[Dict[str, Any]] –

  Latest locations

get_locations

get_locations(device_name: str | None = None, limit: int | None = None, start_dt: datetime | None = None, end_dt: datetime | None = None, order_desc: bool = True) -> List[Dict[str, Any]]

Function to select and return all locations stored in DB

Parameters:

• device_name

  (str | None, default: None ) –

  Device name

• limit

  (int | None, default: None ) –

  Records limit (default := 50)

• start_dt

  (datetime | None, default: None ) –

  Start datetime (default := None)

• end_dt

  (datetime | None, default: None ) –

  End datetime (default := None)

• order_desc

  (bool, default: True ) –

  True if results must be ordered descending, False otherwise (default := True)

Returns:

• List[Dict[str, Any]] –

  List of locations (dict)

insert

insert(data: Dict[str, Any], apply_filters: bool = False, filters: Any | None = None) -> bool

Function to insert a new record into DB

Parameters:

• data

  (Dict[str, Any]) –

  Dictionary of data to insert into DB

• apply_filters

  (bool, default: False ) –

  True to apply filters, False otherwise

• filters

  (Any | None, default: None ) –

  Filters to be applied on data before any insertions

Returns:

• bool –

  True if insert operation was successful, False otherwise