src.digitaltwin.tables

This script contains SQLAlchemy models for various database tables and utility functions for database operations.

Attributes

Base

Classes

`GeospatialLayers`	Class representing the 'geospatial_layers' table.
`UserLogInfo`	Class representing the 'user_log_information' table.

Functions

`create_table`(→ None)	Create a table in the database if it doesn't already exist, using the provided engine.
`check_table_exists`(→ bool)	Check if a table exists in the database.
`execute_query`(→ None)	Execute the given query on the provided engine using a session.

Module Contents

src.digitaltwin.tables.Base

class src.digitaltwin.tables.GeospatialLayers

Bases: Base

Class representing the ‘geospatial_layers’ table.

__tablename__

Name of the database table.

Type:: str

unique_id

Unique identifier for each geospatial layer entry (primary key).

Type:: int

data_provider

Name of the data provider.

Type:: str

layer_id

Identifier for the layer.

Type:: int

table_name

Name of the table containing the data.

Type:: str

unique_column_name

Name of the unique column in the table.

Type:: Optional[str]

coverage_area

Coverage area of the geospatial data, e.g. ‘New Zealand’.

Type:: Optional[str]

url

URL pointing to the geospatial layer.

Type:: str

__tablename__ = 'geospatial_layers'

unique_id

data_provider

layer_id

table_name

unique_column_name

coverage_area

url

__table_args__

class src.digitaltwin.tables.UserLogInfo

Bases: Base

Class representing the ‘user_log_information’ table.

__tablename__

Name of the database table.

Type:: str

unique_id

Unique identifier for each log entry (primary key).

Type:: int

source_table_list

A list of tables (geospatial layers) associated with the log entry.

Type:: Dict[str]

created_at

Timestamp indicating when the log entry was created.

Type:: datetime

geometry

Geometric representation of the catchment area coverage.

Type:: Polygon

__tablename__ = 'user_log_information'

unique_id

source_table_list

created_at

geometry

src.digitaltwin.tables.create_table(engine: sqlalchemy.engine.Engine, table: Base) → None

Create a table in the database if it doesn’t already exist, using the provided engine.

Parameters:

engine (Engine) – The engine used to connect to the database.
table (Base) – Class representing the table to create.

src.digitaltwin.tables.check_table_exists(engine: sqlalchemy.engine.Engine, table_name: str, schema: str = 'public') → bool

Check if a table exists in the database.

Parameters:

engine (Engine) – The engine used to connect to the database.
table_name (str) – The name of the table to check for existence.
schema (str = "public") – The name of the schema where the table resides. Defaults to “public”.

Returns:

True if the table exists, False otherwise.

Return type:

bool

src.digitaltwin.tables.execute_query(engine: sqlalchemy.engine.Engine, query: sqlalchemy.orm.Query) → None

Execute the given query on the provided engine using a session.

Parameters:

engine (Engine) – The engine used to connect to the database.
query (Query) – The query to be executed.

Raises:

Exception – If an error occurs during the execution of the query.