Adds psycopg3 connection pool and make targets for database (#2505)

* Adds psycopg3 connection pool and make targets for database

* fix formating and mypy errors

Author: r-peschke

Makefile

@@ -71,13 +71,28 @@ pip-check:
 coverage:
 	pytest --cov --cov-report html
 
+coverage-show-uncovered:
+	pytest --cov --cov-report term-missing:skip-covered
+
 extract-translations:
 	pybabel extract --no-location --sort-output --omit-header -o openslides_backend/i18n/messages/template-en.pot openslides_backend
 
-db:
+drop-database:
+	make -C global/meta/dev drop-database
+
+create-database:
+	make -C global/meta/dev create-database
+
+apply-db-schema:
+	make -C global/meta/dev apply-db-schema
+	python cli/create_schema.py
+
+create-database-with-schema:
 	make -C global/meta/dev create-database-with-schema
 	python cli/create_schema.py
 
+run-psql:
+	make -C global/meta/dev run-psql
 
 # Build and run production docker container (not usable inside the docker container)
 

cli/create_schema.py

@@ -1,11 +1,9 @@
 from openslides_backend.datastore.shared.postgresql_backend.create_schema import (
     create_schema,
 )
-from openslides_backend.datastore.writer.services import register_services
 
 
 def main() -> None:
-    register_services()
     create_schema()
 
 

dev/entrypoint.sh

@@ -5,6 +5,10 @@ set -e
 source scripts/export_datastore_variables.sh
 scripts/wait.sh $DATASTORE_WRITER_HOST $DATASTORE_WRITER_PORT
 
+printf "\nOpenslides Database:\n"
+python cli/create_schema.py
+printf "\n"
+
 printf "\nMigrations:\n"
 python openslides_backend/migrations/migrate.py finalize
 printf "\n"

openslides_backend/database/db_connection_handling.py

@@ -0,0 +1,61 @@
+import os
+from collections.abc import Callable
+
+import psycopg
+import psycopg_pool
+
+from openslides_backend.shared.env import Environment
+from openslides_backend.shared.exceptions import DatabaseException
+
+env = Environment(os.environ)
+conn_string_without_db = f"host='{env.DATABASE_HOST}' port='{env.DATABASE_PORT}' user='{env.DATABASE_USER}' password='{env.PGPASSWORD}' "
+system_conn_pool = psycopg_pool.ConnectionPool(
+    conninfo=conn_string_without_db + "dbname='postgres'",
+    connection_class=psycopg.Connection,
+    kwargs={"autocommit": True, "row_factory": psycopg.rows.dict_row},
+    min_size=1,
+    max_size=1,
+    open=True,
+    check=psycopg_pool.ConnectionPool.check_connection,
+    name="ConnPool for dev postgres",
+    timeout=5.0,
+    max_waiting=0,
+    max_lifetime=3600.0,
+    max_idle=600.0,
+    reconnect_timeout=300.0,
+    num_workers=1,
+)
+os_conn_pool = psycopg_pool.ConnectionPool(
+    conninfo=conn_string_without_db + f"dbname='{env.DATABASE_NAME}'",
+    connection_class=psycopg.Connection,
+    kwargs={"autocommit": True, "row_factory": psycopg.rows.dict_row},
+    min_size=int(env.DB_POOL_MIN_SIZE),
+    max_size=int(env.DB_POOL_MAX_SIZE),
+    open=False,
+    check=psycopg_pool.ConnectionPool.check_connection,
+    name="ConnPool for openslides-db",
+    timeout=float(env.DB_POOL_TIMEOUT),
+    max_waiting=int(env.DB_POOL_MAX_WAITING),
+    max_lifetime=float(env.DB_POOL_MAX_LIFETIME),
+    max_idle=float(env.DB_POOL_MAX_IDLE),
+    reconnect_timeout=float(env.DB_POOL_RECONNECT_TIMEOUT),
+    num_workers=int(env.DB_POOL_NUM_WORKERS),
+)
+
+
+def get_unpooled_db_connection(
+    db_name: str,
+    autocommit: bool = False,
+    row_factory: Callable = psycopg.rows.dict_row,
+) -> psycopg.Connection:
+    """Use for temporary connections, where pooling is not helpfull like tests and other specific DDL-Connections"""
+    try:
+        db_connection = psycopg.connect(
+            f"host='{env.DATABASE_HOST}' port='{env.DATABASE_PORT}' dbname='{db_name}' user='{env.DATABASE_USER}' password='{env.PGPASSWORD}'",
+            autocommit=autocommit,
+            row_factory=row_factory,
+        )
+        db_connection.isolation_level = psycopg.IsolationLevel.SERIALIZABLE
+    except psycopg.OperationalError as e:
+        raise DatabaseException(f"Cannot connect to postgres: {str(e)}")
+    return db_connection

openslides_backend/datastore/shared/postgresql_backend/create_schema.py

@@ -1,17 +1,52 @@
 import os
 
-from openslides_backend.datastore.shared.di import injector
-from openslides_backend.datastore.shared.postgresql_backend import ConnectionHandler
+from psycopg import Connection, sql
+
+from openslides_backend.database.db_connection_handling import (
+    env,
+    get_unpooled_db_connection,
+)
+from openslides_backend.shared.exceptions import DatabaseException
 
 
 def create_schema() -> None:
     """
     Helper function to write the database schema into the database.
     """
-    connection_handler = injector.get(ConnectionHandler)
-    with connection_handler.get_connection_context():
-        with connection_handler.get_current_connection().cursor() as cursor:
+    connection: Connection
+    try:
+        connection = get_unpooled_db_connection(env.DATABASE_NAME, False)
+    except DatabaseException:
+        conn_postgres = get_unpooled_db_connection("postgres", True)
+        with conn_postgres:
+            with conn_postgres.cursor() as curs:
+                curs.execute(
+                    sql.SQL("CREATE DATABASE {db};").format(
+                        db=sql.Identifier(env.DATABASE_NAME),
+                    )
+                )
+        print("Database openslides created\n")
+        connection = get_unpooled_db_connection(env.DATABASE_NAME, False)
+    with connection:
+        with connection.cursor() as cursor:
+            # idempotent key-value-store schema
             path = os.path.realpath(
                 os.path.join(os.getcwd(), os.path.dirname(__file__), "schema.sql")
             )
-            cursor.execute(open(path).read(), [])
+            cursor.execute(open(path).read())
+            print("Idempotent key-value-schema applied\n")
+
+            # programmatic migrations of schema necessary, only apply if not exists
+            result = cursor.execute(
+                sql.SQL(
+                    "SELECT EXISTS (SELECT FROM pg_tables WHERE tablename = %s AND schemaname = %s);"
+                ),
+                ("organization_t", "public"),
+            ).fetchone()
+            if result and result.get("exists"):
+                return
+            path = os.path.realpath(
+                os.path.join("global", "meta", "dev", "sql", "schema_relational.sql")
+            )
+            cursor.execute(open(path).read())
+            print("Relational schema applied\n")

openslides_backend/shared/env.py

@@ -56,6 +56,21 @@ class Environment(Env):
         "VOTE_PATH": "/internal/vote",
         "VOTE_PORT": "9013",
         "VOTE_PROTOCOL": "http",
+        "DATABASE_HOST": "",
+        "DATABASE_PORT": "5432",
+        "DATABASE_NAME": "openslides",
+        "DATABASE_USER": "openslides",
+        "PGPASSWORD": "openslides",
+        "DATABASE_PASSWORD_FILE": "",
+        # psycopg.ConnectionPool attributes with DB_POOL-prefix
+        "DB_POOL_MIN_SIZE": "4",
+        "DB_POOL_MAX_SIZE": "4",
+        "DB_POOL_TIMEOUT": "30",
+        "DB_POOL_MAX_WAITING": "0",
+        "DB_POOL_MAX_LIFETIME": "3600",
+        "DB_POOL_MAX_IDLE": "60",
+        "DB_POOL_RECONNECT_TIMEOUT": "300",
+        "DB_POOL_NUM_WORKERS": "3",
     }
 
     def __init__(self, os_env: Any, *args: Any, **kwargs: Any) -> None:

openslides_backend/shared/exceptions.py

@@ -84,6 +84,10 @@ class ServiceException(View400Exception):
     pass
 
 
+class DatabaseException(ServiceException):
+    pass
+
+
 class DatastoreException(ServiceException):
     pass