src.digitaltwin.tables ====================== .. py:module:: src.digitaltwin.tables .. autoapi-nested-parse:: This script contains SQLAlchemy models for various database tables and utility functions for database operations. Attributes ---------- .. autoapisummary:: src.digitaltwin.tables.Base Classes ------- .. autoapisummary:: src.digitaltwin.tables.GeospatialLayers src.digitaltwin.tables.UserLogInfo src.digitaltwin.tables.RiverNetworkExclusions src.digitaltwin.tables.RiverNetwork src.digitaltwin.tables.BGFloodModelOutput src.digitaltwin.tables.BuildingFloodStatus Functions --------- .. autoapisummary:: src.digitaltwin.tables.create_table src.digitaltwin.tables.check_table_exists src.digitaltwin.tables.execute_query Module Contents --------------- .. py:data:: Base .. py:class:: GeospatialLayers Bases: :py:obj:`Base` Class representing the 'geospatial_layers' table. .. attribute:: __tablename__ Name of the database table. :type: str .. attribute:: unique_id Unique identifier for each geospatial layer entry (primary key). :type: int .. attribute:: data_provider Name of the data provider. :type: str .. attribute:: layer_id Identifier for the layer. :type: int .. attribute:: table_name Name of the table containing the data. :type: str .. attribute:: unique_column_name Name of the unique column in the table. :type: Optional[str] .. attribute:: coverage_area Coverage area of the geospatial data, e.g. 'New Zealand'. :type: Optional[str] .. attribute:: url URL pointing to the geospatial layer. :type: str .. py:attribute:: __tablename__ :value: 'geospatial_layers' .. py:attribute:: unique_id .. py:attribute:: data_provider .. py:attribute:: layer_id .. py:attribute:: table_name .. py:attribute:: unique_column_name .. py:attribute:: coverage_area .. py:attribute:: url .. py:attribute:: __table_args__ .. py:class:: UserLogInfo Bases: :py:obj:`Base` Class representing the 'user_log_information' table. .. attribute:: __tablename__ Name of the database table. :type: str .. attribute:: unique_id Unique identifier for each log entry (primary key). :type: int .. attribute:: source_table_list A list of tables (geospatial layers) associated with the log entry. :type: Dict[str] .. attribute:: created_at Timestamp indicating when the log entry was created. :type: datetime .. attribute:: geometry Geometric representation of the catchment area coverage. :type: Polygon .. py:attribute:: __tablename__ :value: 'user_log_information' .. py:attribute:: unique_id .. py:attribute:: source_table_list .. py:attribute:: created_at .. py:attribute:: geometry .. py:class:: RiverNetworkExclusions Bases: :py:obj:`Base` Class representing the 'rec_network_exclusions' table. .. attribute:: __tablename__ Name of the database table. :type: str .. attribute:: rec_network_id An identifier for the river network associated with each new run. :type: int .. attribute:: objectid An identifier for the REC river object matching from the 'rec_data' table. :type: int .. attribute:: exclusion_cause Cause of exclusion, i.e., the reason why the REC river geometry was excluded. :type: str .. attribute:: geometry Geometric representation of the excluded REC river features. :type: LineString .. py:attribute:: __tablename__ :value: 'rec_network_exclusions' .. py:attribute:: rec_network_id .. py:attribute:: objectid .. py:attribute:: exclusion_cause .. py:attribute:: geometry .. py:attribute:: __table_args__ .. py:class:: RiverNetwork Bases: :py:obj:`Base` Class representing the 'rec_network' table. .. attribute:: __tablename__ Name of the database table. :type: str .. attribute:: rec_network_id An identifier for the river network associated with each new run (primary key). :type: int .. attribute:: network_path Path to the REC river network file. :type: str .. attribute:: network_data_path Path to the REC river network data file for the AOI. :type: str .. attribute:: created_at Timestamp indicating when the output was created. :type: datetime .. attribute:: geometry Geometric representation of the catchment area coverage. :type: Polygon .. py:attribute:: __tablename__ :value: 'rec_network' .. py:attribute:: rec_network_id .. py:attribute:: network_path .. py:attribute:: network_data_path .. py:attribute:: created_at .. py:attribute:: geometry .. py:class:: BGFloodModelOutput Bases: :py:obj:`Base` Class representing the 'bg_flood_model_output' table. .. attribute:: __tablename__ Name of the database table. :type: str .. attribute:: unique_id Unique identifier for each entry (primary key). :type: int .. attribute:: file_name Name of the flood model output file. :type: str .. attribute:: file_path Path to the flood model output file. :type: str .. attribute:: created_at Timestamp indicating when the output was created. :type: datetime .. attribute:: geometry Geometric representation of the catchment area coverage. :type: Polygon .. py:attribute:: __tablename__ :value: 'bg_flood_model_output' .. py:attribute:: unique_id .. py:attribute:: file_name .. py:attribute:: file_path .. py:attribute:: created_at .. py:attribute:: geometry .. py:class:: BuildingFloodStatus Bases: :py:obj:`Base` Class representing the 'building_flood_status' table. Represents if a building is flooded for a given flood model output .. attribute:: __tablename__ Name of the database table. :type: str .. attribute:: unique_id Unique identifier for each entry (primary key). :type: int .. attribute:: building_outline_id Foreign key building outline id matching from nz_building_outlines table :type: int .. attribute:: is_flooded If the building is flooded or not :type: bool .. attribute:: flood_model_id Foreign key mathing the unique_id from bg_flood_model_output table :type: int. .. py:attribute:: __tablename__ :value: 'building_flood_status' .. py:attribute:: unique_id .. py:attribute:: building_outline_id .. py:attribute:: is_flooded .. py:attribute:: flood_model_id .. py:function:: 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. :param engine: The engine used to connect to the database. :type engine: Engine :param table: Class representing the table to create. :type table: Base :returns: This function does not return any value. :rtype: None .. py:function:: check_table_exists(engine: sqlalchemy.engine.Engine, table_name: str, schema: str = 'public') -> bool Check if a table exists in the database. :param engine: The engine used to connect to the database. :type engine: Engine :param table_name: The name of the table to check for existence. :type table_name: str :param schema: The name of the schema where the table resides. Defaults to "public". :type schema: str = "public" :returns: True if the table exists, False otherwise. :rtype: bool .. py:function:: execute_query(engine: sqlalchemy.engine.Engine, query) -> None Execute the given query on the provided engine using a session. :param engine: The engine used to connect to the database. :type engine: Engine :param query: The query to be executed. :returns: This function does not return any value. :rtype: None :raises Exception: If an error occurs during the execution of the query.