script_helper.py

# helferscript mit klassen für den URI und adressen update
# schuldatenbank heißt in der DB "schulen"
import requests
import psycopg2
import json


class DBTools:
    """
    Tools für das arbeiten mit der Datenbank, hier sind die select Funktionen
    zum auslesen der Daten und zum updaten der Daten mit den Adressen.
    """
    def __init__(self, conn: str) -> None:
        """
        Aufbauen der Verbindung zu der Datenbank

        :param str data: Connection String mit allen Informationen für die
            Verbindung zur Datenbank
        """
        self.db_connection = psycopg2.connect(conn)
        self.cursor = self.db_connection.cursor()

    def select_adress(self) -> "list[tuple[str, ... ]]":
        """
        Funktion zum selektieren der Adressdaten der Schulen aus der Datenbank.
        Die folgenden Daten werden aus der Datenbank ausgelesen:
        * ID der Schule in der Datenbank
        * Hausnummer
        * Straßenname
        * PLZ
        * Ortsname

        :returns: Liste mit mehreren Tupeln welche die Adressdaten der
            Schulen enthalten.
        :rtype: list
        """

        select_sql = """
        select
            lokaleID,
            adressealt_hnr,
            adressealt_strasse,
            adressealt_plz,
            adressealt_ort
        from
            prototyp_areg.schulen;"""

        self.cursor.execute(select_sql)
        query_result = self.cursor.fetchall()

        return query_result

    def select_uri(self) -> "list[tuple[str, ... ]]":
        """
        Funktion zum selektieren der URI aus dem Datensatz um diese mit
        aktuellen AREG Adressdaten anzureichern.

        :return: Liste mit meheren Tupeln welche die AREG URI enthalten.
            Es wird sichergestellt, dass keine NULL Werte vorkommen.
        :rtype: list
        """

        select_sql = """
        select
            distinct(areg_uri)
        from
            prototyp_areg.schulen
        where
            areg_uri like 'https://%';"""

        self.cursor.execute(select_sql)
        query_data = self.cursor.fetchall()

        return query_data

    def update_uri(self, param_dbid: str, param_uri: str) -> None:
        """
        Funktion zum Update der Datenbank mit den neuen URI aus dem Geocoder.

        :param str param_dbid: ID der Schule aus der Datenbank (nicht AREG URI)
        :param str param_uri: AREG URI
        """

        sql_alter = """
        alter table
            prototyp_areg.schulen
        add column if not exists
            areg_uri varchar;
        """

        sql_update = f"""
        update
            prototyp_areg.schulen
        set
            areg_uri='{param_uri}'
        where
            lokaleID='{param_dbid}';
        """

        self.cursor.execute(sql_alter)
        self.cursor.execute(sql_update)

    def update_adress(self, data: "dict[str,str]", param_uri: str) -> None:
        """
        Funktion zum Update der Datenbank mit den neuen Adressdaten aus dem
        AREG anhand der AREG URI.

        :param dict data: Dictionary mit neuen Adressdaten aus der AREG API
        :param str param_uri: AREG URI zum zuordnen der Adressdaten
        """
        geometry = str(data['geometry'])
        geometry = geometry.replace("'", '"')

        sql_alter = """
        alter table
            prototyp_areg.schulen
        add column if not exists
            hnr varchar,
        add column if not exists
            strasse varchar,
        add column if not exists
            plz varchar,
        add column if not exists
            ort varchar,
        add column if not exists
            geom geometry(Point, 4326);
        """

        sql_update = f"""
        update
            prototyp_areg.schulen
        set
            hnr='{data['hnr']}',
            strasse='{data['stn']}',
            plz='{data['plz']}',
            ort='{data['ort']}',
            geom=ST_GeomFromGeoJSON('{geometry}')
        where
            areg_uri='{param_uri}';
        """

        self.cursor.execute(sql_alter)
        self.cursor.execute(sql_update)

    def closecursor(self) -> None:
        """
        Funktion zum Commit und Schießen der Datenbankverbindung nachdem alle
        Änderungen vorgenommen wurden.
        """
        # schließt den cursor
        self.db_connection.commit()
        self.cursor.close()
        self.db_connection.close()


class APITools:
    """
    Tools für das Arbeiten mit der pygeoAPI. Hier sind die Funktionen zum
    aufrufen des Geocoder Prozesses und zum Finden der Adresse anhand der URI
    """
    def __init__(self, param_url):
        """
        Festlegen der Basis URL der API für die weitere Verwendung in den
        Funktionen der Klasse.

        :param str param_url: Basis URL der API
        """
        self.base_url = param_url

    def geocode(self, data: "dict[str,str]") -> "dict[str,str]":
        """
        Funktion zum Aufrufen des Geocoder Prozesses in der API. Bereinigung
        der Adressdaten finden nicht in dieser Funktion oder diesem Skript
        sondern im Geocoder statt!

        :param dict data: Dictionary mit den alten Adressdaten
        :returns: Dictionary entweder mit einem `uri` Objekt oder einem
            `message` Objekt. Wenn `uri` kommt ist alles gut, wenn `message`
            kommt ist ein Fehler aufgetreten. Der genaue Fehler ist in
            `message` enthalten.
        :rtype: dict
        """
        url = self.base_url + r"processes/geocoder/execution"

        result = {"message": "you should not see this"}
        request_body = {
            "inputs": {
                "hnr": data["hnr"],
                "ort": data["ort"],
                "plz": data["plz"],
                "stn": data["stn"],
                "uri": True
            }
        }

        try:
            api_result = requests.post(url, json.dumps(request_body))
            api_result.raise_for_status()
            result = api_result.json()
        except requests.exceptions.ConnectionError as e:
            result = {"message": str(e)}
        except requests.exceptions.Timeout as e:
            result = {"message": str(e)}
        except requests.exceptions.HTTPError as e:
            result = {"message": str(e)}

        return result

    def findadresse(self, param_uri: str) -> "dict[str,str]":
        """
        Funktion zum finden der Adresse im AREG anhand der vergebenen AREG URI.

        :param str param_uri: URL zur API mit AREG URI der Adresse.

        :return: Adressdaten des jeweiligen Items das der AREG URI entspricht
        :rtype: dict
        """
        url = param_uri + r"?f=json"
        result = {}
        response = requests.get(url)
        response_json = response.json()

        result['hnr'] = response_json['properties']['hnr']
        result['stn'] = response_json['properties']['stn']
        result['plz'] = response_json['properties']['plz']
        result['ort'] = response_json['properties']['ort']
        result['geometry'] = response_json['geometry']

        return result