Implement database classes

nrw/database/__init__.py

@@ -0,0 +1 @@
+"""Die Datenbankklassen nach den Vorgaben des Landes NRW."""

nrw/database/_query_result.py

@@ -0,0 +1,63 @@
+"""Klasse `QueryResult`."""
+
+from __future__ import annotations
+
+__all__: Final[tuple[str]] = ("QueryResult",)
+
+from typing import Any, Final
+
+
+class QueryResult:
+    """Ein Objekt der Klasse `QueryResult` stellt die Ergebnistabelle einer
+    Datenbankanfrage mit Hilfe
+    der Klasse `DatabaseConnector` dar. Objekte dieser Klasse werden nur von der
+    Klasse `DatabaseConnector` erstellt.
+    Die Klasse verfügt über keinen öffentlichen Konstruktor.
+    """
+
+    __slots__: Final[tuple[str, str, str]] = ("_data", "_column_names", "_column_types")
+
+    def __init__(
+        self,
+        data: list[tuple[Any, ...]],
+        column_names: tuple[str, ...],
+        column_types: tuple[Any, ...],
+    ) -> None:
+        """Interner Konstruktor."""
+        self._data: list[tuple[Any, ...]] = data
+        self._column_names: tuple[str, ...] = column_names
+        self._column_types: tuple[Any, ...] = column_types
+
+    @property
+    def data(self) -> list[tuple[Any, ...]]:
+        """Die Anfrage liefert die Einträge der Ergebnistabelle als eine `list`
+        welche wiederum `tuple` enthält. Der erste Index stellt die Zeile und der zweite
+        die Spalte dar (d.h. Object[zeile][spalte]).
+        """
+        return self._data
+
+    @property
+    def column_names(self) -> tuple[str, ...]:
+        """Die Anfrage liefert die Bezeichner der Spalten der Ergebnistabelle als
+        `tuple` vom Typ `str` zurück.
+        """
+        return self._column_names
+
+    @property
+    def column_types(self) -> tuple[Any, ...]:
+        """Die Anfrage liefert (wenn möglich) die Typenbezeichnung der Spalten der
+        Ergebnistabelle als `tuple` vom jeweiligen Typ zurück.
+        Die Bezeichnungen entsprechen den Angaben in der `MySQL`-Datenbank.
+        """
+        return self._column_types
+
+    @property
+    def row_count(self) -> int:
+        """Die Anfrage liefert die Anzahl der Zeilen der Ergebnistabelle als `int`."""
+        return len(self._data) if self._data is not None else 0
+
+    @property
+    def column_count(self) -> int:
+        """Die Anfrage liefert die Anzahl der Spalten der Ergebnistabelle als `int`."""
+        assert all(len(self._data[0]) == len(data) for data in self._data[1:])
+        return len(self._data[0])

nrw/database/_query_result.pyi

@@ -0,0 +1,22 @@
+__all__: Final[tuple[str]] = ("QueryResult",)
+
+from typing import Any, Final
+
+class QueryResult:
+    __slots__: Final[tuple[str, str, str]] = ("_data", "_column_names", "_column_types")
+    def __init__(
+        self,
+        data: list[tuple[Any, ...]],
+        column_names: tuple[str, ...],
+        column_types: tuple[Any, ...],
+    ) -> None: ...
+    @property
+    def data(self) -> list[tuple[Any, ...]]: ...
+    @property
+    def column_names(self) -> tuple[str, ...]: ...
+    @property
+    def column_types(self) -> tuple[Any, ...]: ...
+    @property
+    def row_count(self) -> int: ...
+    @property
+    def column_count(self) -> int: ...

nrw/database/msaccess.py

@@ -0,0 +1,119 @@
+"""Klasse `DatabaseConnector`."""
+
+from __future__ import annotations
+
+__all__: Final[tuple[str]] = ("DatabaseConnector",)
+
+from typing import Final
+
+import pyodbc  # type: ignore[import-not-found]
+
+from nrw.database._query_result import QueryResult
+
+
+class DatabaseConnector:
+    """Ein Objekt der Klasse `DatabaseConnector` ermöglicht die Abfrage und Manipulation
+    einer `MSAccess`-Datenbank.
+    Beim Erzeugen des Objekts wird eine Datenbankverbindung aufgebaut, so dass
+    anschließend SQL-Anweisungen an diese Datenbank gerichtet werden können.
+    """
+
+    __slots__: Final[tuple[str, str, str]] = (
+        "_connection",
+        "_current_query_result",
+        "_message",
+    )
+
+    # pylint: disable=W0613, W0718
+
+    def __init__(
+        self,
+        ip: None,
+        port: None,
+        database: str,
+        username: None,
+        password: None,
+    ) -> None:
+        """Ein Objekt vom Typ `DatabaseConnector` wird erstellt, und eine Verbindung zur
+        Datenbank wird aufgebaut. Mit den Parametern `ip` und `port` werden die
+        IP-Adresse und die Port-Nummer übergeben, unter denen die Datenbank mit Namen
+        `database` zu erreichen ist.
+        Mit den Parametern `username` und `password` werden Benutzername und Passwort
+        für die Datenbank übergeben.
+
+        Für `MSAccess` wird nur `database` benötigt.
+        """
+        self._current_query_result: QueryResult | None = None
+        self._message: str | None = None
+
+        try:
+            self._connection = pyodbc.connect(
+                f"Driver={{Microsoft Access Driver (*.mdb, *.accdb)}};DBQ={database};",
+                autocommit=True,
+            )
+        except Exception as exception:
+            self._connection = None
+            self._message = str(exception)
+
+    def execute_statement(self, sql_statement: str) -> None:
+        """Der Auftrag schickt den im Parameter `sql_statement` enthaltenen SQL-Befehl
+        an die Datenbank ab.
+        Handelt es sich bei `sql_statement` um einen SQL-Befehl, der eine Ergebnismenge
+        liefert, so kann dieses Ergebnis anschließend mit dem Property
+        `current_query_result` abgerufen werden.
+        """
+        self._current_query_result = None
+        self._message = None
+
+        if self._connection is None:
+            self._message = "No connection"
+            return
+
+        try:
+            with self._connection.cursor() as cursor:
+                cursor.execute(sql_statement)
+                if data := cursor.fetchall():
+                    assert cursor.description is not None, "No description"
+                    column_names: tuple[str, ...] = tuple(
+                        column[0] for column in cursor.description
+                    )
+                    colum_types: tuple[int, ...] = tuple(
+                        column[1] for column in cursor.description
+                    )
+                    self._current_query_result = QueryResult(
+                        [tuple(row) for row in data],
+                        column_names,
+                        colum_types,
+                    )
+        except Exception as exception:
+            if str(exception) != "No results.  Previous SQL was not a query.":
+                self._message = str(exception)
+
+    @property
+    def current_query_result(self) -> QueryResult | None:
+        """Die Anfrage liefert das Ergebnis des letzten mit der Methode
+        `execute_statement` an die Datenbank geschickten SQL-Befehls als Objekt vom Typ
+        `QueryResult` zurück.
+        Wurde bisher kein SQL-Befehl abgeschickt oder ergab der letzte Aufruf von
+        `execute_statement` keine Ergebnismenge (z.B. bei einem INSERT-Befehl oder einem
+        Syntaxfehler), so wird `None` geliefert.
+        """
+        return self._current_query_result
+
+    @property
+    def error_message(self) -> str | None:
+        """Die Anfrage liefert `None` oder eine Fehlermeldung, die sich jeweils auf die
+        letzte zuvor ausgefuehrte Datenbankoperation bezieht.
+        """
+        return self._message
+
+    def close(self) -> None:
+        """Die Datenbankverbindung wird geschlossen."""
+        if self._connection is None:
+            self._message = "No connection"
+            return
+
+        try:
+            self._connection.close()
+        except Exception as exception:
+            self._message = str(exception)

nrw/database/msaccess.pyi

@@ -0,0 +1,27 @@
+__all__: Final[tuple[str]] = ("DatabaseConnector",)
+
+from typing import Final
+
+from nrw.database._query_result import QueryResult
+
+class DatabaseConnector:
+    __slots__: Final[tuple[str, str, str]] = (
+        "_connection",
+        "_current_query_result",
+        "_message",
+    )
+
+    def __init__(
+        self,
+        ip: None,
+        port: None,
+        database: str,
+        username: None,
+        password: None,
+    ) -> None: ...
+    def execute_statement(self, sql_statement: str) -> None: ...
+    @property
+    def current_query_result(self) -> QueryResult | None: ...
+    @property
+    def error_message(self) -> str | None: ...
+    def close(self) -> None: ...

nrw/database/mysql.py

@@ -0,0 +1,126 @@
+"""Klasse `DatabaseConnector`."""
+
+from __future__ import annotations
+
+__all__: Final[tuple[str]] = ("DatabaseConnector",)
+
+from typing import TYPE_CHECKING, Final
+
+import mysql.connector
+
+from nrw.database._query_result import QueryResult
+
+if TYPE_CHECKING:
+    from mysql.connector.abstracts import MySQLConnectionAbstract
+    from mysql.connector.pooling import PooledMySQLConnection
+
+
+class DatabaseConnector:
+    """Ein Objekt der Klasse `DatabaseConnector` ermöglicht die Abfrage und Manipulation
+    einer `MySQL`-Datenbank.
+    Beim Erzeugen des Objekts wird eine Datenbankverbindung aufgebaut, so dass
+    anschließend SQL-Anweisungen an diese Datenbank gerichtet werden können.
+    """
+
+    # pylint: disable=W0613, W0718, R0913
+
+    __slots__: Final[tuple[str, str, str]] = (
+        "_connection",
+        "_current_query_result",
+        "_message",
+    )
+
+    def __init__(
+        self,
+        ip: str,
+        port: int,
+        database: str,
+        username: str,
+        password: str,
+    ) -> None:
+        """Ein Objekt vom Typ `DatabaseConnector` wird erstellt, und eine Verbindung zur
+        Datenbank wird aufgebaut. Mit den Parametern `ip` und `port` werden die
+        IP-Adresse und die Port-Nummer übergeben, unter denen die Datenbank mit Namen
+        `database` zu erreichen ist.
+        Mit den Parametern `username` und `password` werden Benutzername und Passwort
+        für die Datenbank übergeben.
+        """
+        self._current_query_result: QueryResult | None = None
+        self._message: str | None = None
+
+        try:
+            self._connection: PooledMySQLConnection | MySQLConnectionAbstract | None = (
+                mysql.connector.connect(
+                    user=username,
+                    password=password,
+                    host=ip,
+                    port=port,
+                    database=database,
+                    autocommit=True,
+                )
+            )
+        except Exception as exception:
+            self._connection = None
+            self._message = str(exception)
+
+    def execute_statement(self, sql_statement: str) -> None:
+        """Der Auftrag schickt den im Parameter `sql_statement` enthaltenen SQL-Befehl
+        an die Datenbank ab.
+        Handelt es sich bei `sql_statement` um einen SQL-Befehl, der eine Ergebnismenge
+        liefert, so kann dieses Ergebnis anschließend mit dem Property
+        `current_query_result` abgerufen werden.
+        """
+        self._current_query_result = None
+        self._message = None
+
+        if self._connection is None:
+            self._message = "No connection"
+            return
+
+        try:
+            with self._connection.cursor(dictionary=False) as cursor:
+                cursor.execute(sql_statement)
+                if data := cursor.fetchall():
+                    assert cursor.description is not None, "No description"
+                    column_names: tuple[str, ...] = tuple(
+                        column[0] for column in cursor.description
+                    )
+                    colum_types: tuple[int, ...] = tuple(
+                        column[1] for column in cursor.description
+                    )
+                    self._current_query_result = QueryResult(
+                        data,  # type: ignore[arg-type]
+                        column_names,
+                        colum_types,
+                    )
+        except Exception as exception:
+            self._message = str(exception)
+
+    @property
+    def current_query_result(self) -> QueryResult | None:
+        """Die Anfrage liefert das Ergebnis des letzten mit der Methode
+        `execute_statement` an die Datenbank geschickten SQL-Befehls als Objekt vom Typ
+        `QueryResult` zurück.
+        Wurde bisher kein SQL-Befehl abgeschickt oder ergab der letzte Aufruf von
+        `execute_statement` keine Ergebnismenge (z.B. bei einem INSERT-Befehl oder einem
+        Syntaxfehler), so wird `None` geliefert.
+        """
+        return self._current_query_result
+
+    @property
+    def error_message(self) -> str | None:
+        """Die Anfrage liefert `None` oder eine Fehlermeldung, die sich jeweils auf die
+        letzte zuvor ausgefuehrte Datenbankoperation bezieht.
+        """
+        return self._message
+
+    def close(self) -> None:
+        """Die Datenbankverbindung wird geschlossen."""
+        if self._connection is None:
+            self._message = "No connection"
+            return
+
+        try:
+            self._connection.close()
+        except Exception as exception:
+            self._message = str(exception)

nrw/database/mysql.pyi

@@ -0,0 +1,28 @@
+__all__: Final[tuple[str]] = ("DatabaseConnector",)
+
+from typing import Final
+
+from nrw.database._query_result import QueryResult
+
+class DatabaseConnector:
+
+    __slots__: Final[tuple[str, str, str]] = (
+        "_connection",
+        "_current_query_result",
+        "_message",
+    )
+
+    def __init__(
+        self,
+        ip: str,
+        port: int,
+        database: str,
+        username: str,
+        password: str,
+    ) -> None: ...
+    def execute_statement(self, sql_statement: str) -> None: ...
+    @property
+    def current_query_result(self) -> QueryResult | None: ...
+    @property
+    def error_message(self) -> str | None: ...
+    def close(self) -> None: ...