use python-sql instead of psycopg.sql for generating sql-queries

dev/Makefile

@@ -1,4 +1,5 @@
 # Commands inside the container
+paths = src/ tests/
 
 all: pyupgrade black autoflake isort flake8 mypy
 
@@ -9,22 +10,22 @@ check-pyupgrade:
 	pyupgrade --py310-plus $$(find . -name '*.py')
 
 black:
-	black src/
+	black $(paths)
 
 check-black:
-	black --check --diff src/
+	black --check --diff $(paths)
 
 autoflake:
-	autoflake src/
+	autoflake $(paths)
 
 isort:
-	isort src/
+	isort $(paths)
 
 check-isort:
-	isort --check-only --diff src/
+	isort --check-only --diff $(paths)
 
 flake8:
-	flake8 src/
+	flake8 $(paths)
 
 mypy:
 	mypy src/

dev/setup.cfg

@@ -17,3 +17,12 @@ extend-ignore = E203,E501
 
 [mypy]
 disallow_untyped_defs = true
+
+[mypy-sql.*]
+ignore_missing_imports = true
+
+[mypy-python_sql.*]
+ignore_missing_imports = true
+
+[mypy-helper_get_names.*]
+ignore_missing_imports = true

dev/sql/schema_relational.sql

@@ -55,7 +55,7 @@ BEGIN
 END;
 $$ LANGUAGE plpgsql;
 
--- MODELS_YML_CHECKSUM = '959d3a581a8015a294d769587ebb1b6e'
+-- MODELS_YML_CHECKSUM = '385e6beb7945826015352886f2d0f97b'
 -- Type definitions
 
 -- Table definitions

dev/src/db_utils.py

@@ -1,83 +1,38 @@
-from typing import Any, cast
+from typing import Any
 
-from psycopg import Cursor, sql
+from src.python_sql import Column, Table
 
 
 class DbUtils:
     @classmethod
-    def get_pg_array_for_cu(cls, data: list) -> str:
-        return f"""{{"{'","'.join(item for item in data)}"}}"""
-
-    @classmethod
-    def insert_wrapper(
-        cls, curs: Cursor, table_name: str, data: dict[str, Any]
-    ) -> None | int:
-        query = f"INSERT INTO {table_name} ({', '.join(data.keys())}) VALUES({{}}) RETURNING id;"
-        query = (
-            sql.SQL(query)
-            .format(
-                sql.SQL(", ").join(sql.Placeholder() * len(data.keys())),
-            )
-            .as_string(curs)
-        )
-        result = curs.execute(query, tuple(data.values())).fetchone()
-        if isinstance(result, dict):
-            return result.get("id", 0)
-        return None
-
-    @classmethod
-    def insert_many_wrapper(
+    def get_columns_and_values_for_insert(
         cls,
-        curs: Cursor,
-        table_name: str,
+        table: Table,
         data_list: list[dict[str, Any]],
-        returning: str = "id",
-    ) -> list[int]:
-        ids: list[int] = []
+    ) -> tuple[list[Column], list[list[dict[str, Any]]]]:
+        """
+        takes a list of dicts, each one to be inserted
+        Takes care of columns and row positions and fills
+        not existent columns in row with "None"
+        """
+        columns: list[Column] = []
+        values: list[list[dict[str, Any]]] = []
         if not data_list:
-            return ids
+            return columns, values
         # use all keys in same sequence
         keys_set: set = set()
         for data in data_list:
             keys_set.update(data.keys())
         keys: list = sorted(keys_set)
-        temp_data = {k: None for k in keys}
+        columns = [Column(table, key) for key in keys]
+        values = [[row.get(k, None) for k in keys] for row in data_list]
+        return columns, values
 
-        dates = [temp_data | data for data in data_list]
-        query = f"INSERT INTO {table_name} ({', '.join(keys)}) VALUES({{}}){' RETURNING ' + returning if returning else ''};"
-        query = (
-            sql.SQL(query)
-            .format(
-                sql.SQL(", ").join(sql.Placeholder() * len(keys)),
-            )
-            .as_string(curs)
-        )
-        curs.executemany(
-            query,
-            tuple(tuple(v for _, v in sorted(data.items())) for data in dates),
-            returning=bool(returning),
-        )
-        ids = []
-        if returning:
-            while True:
-                ids.append(cast(dict, curs.fetchone())[returning])
-                if not curs.nextset():
-                    break
-        return ids
+    @classmethod
+    def get_columns_from_list(cls, table: Table, items: list[str]) -> list[Column]:
+        return [Column(table, item) for item in items]
 
     @classmethod
-    def select_id_wrapper(
-        cls,
-        curs: Cursor,
-        table_name: str,
-        id_: int | None = None,
-        field_names: list[str] = [],
-    ) -> dict[str, Any] | list[dict[str, Any]]:
-        """select with single id or all for fields in list or all fields"""
-        query = sql.SQL(
-            f"SELECT {', '.join(field_names) if field_names else '*'} FROM {table_name}{' where id = %s' if id_ else ''}"
-        )
-        if id_:
-            return result if (result := curs.execute(query, (id_,)).fetchone()) else {}
-        else:
-            return curs.execute(query).fetchall()
+    def get_pg_array_for_cu(cls, data: list) -> str:
+        """converts a value list into string used for complete array field"""
+        return f"""{{"{'","'.join(item for item in data)}"}}"""

dev/src/generate_sql_schema.py

@@ -1060,7 +1060,6 @@ def generate_field_or_sql_decision(
             ("1t", "1rR"): (FieldSqlErrorType.SQL, False),
             ("1tR", "1Gr"): (FieldSqlErrorType.SQL, False),
             ("1tR", "1GrR"): (FieldSqlErrorType.SQL, False),
-            ("1rR", "1t"): (FieldSqlErrorType.FIELD, False),
             ("nGt", "nt"): (FieldSqlErrorType.SQL, True),
             ("nr", ""): (FieldSqlErrorType.SQL, True),
             ("nt", "1Gr"): (FieldSqlErrorType.SQL, False),

dev/src/python_sql.py

@@ -0,0 +1,2 @@
+# imports from sql for usage with type: ignore, because mypy don'T recognize the original import
+from sql import Column, Table  # type: ignore  # noqa:F401