Skip to content
Snippets Groups Projects
obm_buildings_upsert.py 2.75 KiB
Newer Older
class ObmBuildingsUpsert:
    def __call__(
        self,
        database,
        geometry,
        osm_id,
        floorspace=None,
        occupancy=None,
        stories=None,
        relation_id=None,
        quadkey=None,
        *args,
        **kwargs,
    ):
        """
        Insert or update a building in the `obm_buildings` table.

        Args:
            database (PostGISDatabase):
                Exposure database.
            geometry (str):
                WKT formatted geometry of the building, wrapped in the `ST_GeomFromText()`
                function.
            osm_id (int):
                OSM ID of the building.
            floorspace (float):
                The calculated floorspace of the building.
            occupancy (str):
                The occupancy type formatted to the GEM occupancy types.
            stories (int):
                Number of stories in the building.
            relation_id (int):
                ID of the OSM relation feature that this building is part of.
            quadkey (str):
                Quadkey of the tile that the building is located in.
        """

        sql_statement = f"""
            INSERT INTO obm_buildings
                (osm_id, geometry, floorspace, occupancy, storeys, relation_id, quadkey,
                last_update)
            VALUES (
                {osm_id},
                {geometry},
                {self.format_value(floorspace)},
                {self.format_value(occupancy)},
                {self.format_value(stories)},
                {self.format_value(relation_id)},
                {self.format_value(quadkey)},
                NOW()
                )
            ON CONFLICT (osm_id) DO UPDATE
                SET geometry = excluded.geometry,
                    floorspace = excluded.floorspace,
                    occupancy = excluded.occupancy,
                    storeys = excluded.storeys,
                    relation_id = excluded.relation_id,
                    quadkey = excluded.quadkey,
                    last_update = excluded.last_update
            """

        database.cursor.execute(sql_statement)
        database.connection.commit()

    @staticmethod
    def format_value(value):
        """
        Format values correctly to be used in an SQL statement.

        Args:
            value (object):
                The value that needs to be formatted.

        Returns:
            Correctly formatted value.
        """

        # None values should be `NULL` in PostgreSQL
        if value is None:
            return "NULL"
        # If a value is a string, it should be in quotation marks
        elif isinstance(value, str):
            return f"'{value}'"
        # Else return a string representation of the value
        else:
            return str(value)