Added a field_validator to the config.py that will url encode (#79)

Added a field_validator to the config.py that will url encode postgres
connection string

to translate special characters to url encoding that will not cause the
pydantic test to fail.

Added a pytest function to test the encoding

pyproject.toml

@@ -47,8 +47,11 @@ addopts = [
 ]
 testpaths = ["tests"]
 env = [
-  "INST_CONN=postgresql+asyncpg://localhost",
   "INST_DB_SCHEMA=main",
+  "INST_DB_USER=user",
+  "INST_DB_PWD=user",
+  "INST_DB_HOST=localhost:5432",
+  "INST_DB_NAME=fi",
   "KC_URL=http://localhost",
   "KC_REALM=",
   "KC_ADMIN_CLIENT_ID=",

src/.env.local

@@ -13,7 +13,6 @@ INST_DB_USER=fi
 INST_DB_PWD=fi
 INST_DB_HOST=localhost:5432
 INST_DB_SCHEMA=public
-INST_CONN=postgresql+asyncpg://${INST_DB_USER}:${INST_DB_PWD}@${INST_DB_HOST}/${INST_DB_NAME}
 JWT_OPTS_VERIFY_AT_HASH="false"
 JWT_OPTS_VERIFY_AUD="false"
 JWT_OPTS_VERIFY_ISS="false"

src/.env.template

@@ -13,4 +13,4 @@ INST_DB_USER=
 INST_DB_PWD=
 INST_DB_HOST=
 INST_DB_SCHEMA=
-INST_CONN=postgresql+asyncpg://${INST_DB_USER}:${INST_DB_PWD}@${INST_DB_HOST}/${INST_DB_NAME}
+# INST_DB_SCHEME This can be added if needing to override the default of 'postgresql+asyncpg'

src/config.py

@@ -1,7 +1,8 @@
 import os
+from urllib import parse
 from typing import Dict, Any
 
-from pydantic import TypeAdapter
+from pydantic import TypeAdapter, field_validator, ValidationInfo
 from pydantic.networks import HttpUrl, PostgresDsn
 from pydantic.types import SecretStr
 from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -14,8 +15,13 @@
 
 
 class Settings(BaseSettings):
-    inst_conn: PostgresDsn
     inst_db_schema: str = "public"
+    inst_db_name: str
+    inst_db_user: str
+    inst_db_pwd: str
+    inst_db_host: str
+    inst_db_scheme: str = "postgresql+asyncpg"
+    inst_conn: PostgresDsn | None = None
     auth_client: str
     auth_url: HttpUrl
     token_url: HttpUrl
@@ -31,6 +37,18 @@ def __init__(self, **data):
         super().__init__(**data)
         self.set_jwt_opts()
 
+    @field_validator("inst_conn", mode="before")
+    @classmethod
+    def build_postgres_dsn(cls, postgres_dsn, info: ValidationInfo) -> Any:
+        postgres_dsn = PostgresDsn.build(
+            scheme=info.data.get("inst_db_scheme"),
+            username=info.data.get("inst_db_user"),
+            password=parse.quote(info.data.get("inst_db_pwd"), safe=""),
+            host=info.data.get("inst_db_host"),
+            path=info.data.get("inst_db_name"),
+        )
+        return str(postgres_dsn)
+
     def set_jwt_opts(self) -> None:
         """
         Converts `jwt_opts_` prefixed settings, and env vars into JWT options dictionary.

tests/app/test_config.py

@@ -2,6 +2,18 @@
 from config import Settings
 
 
+def test_postgres_dsn_building():
+    mock_config = {
+        "inst_db_name": "test",
+        "inst_db_user": "user",
+        "inst_db_pwd": "\\z9-/tgb76#@",
+        "inst_db_host": "test:5432",
+        "inst_db_scehma": "test",
+    }
+    settings = Settings(**mock_config)
+    assert str(settings.inst_conn) == "postgresql+asyncpg://user:%5Cz9-%2Ftgb76%23%40@test:5432/test"
+
+
 def test_jwt_opts_valid_values():
     mock_config = {
         "jwt_opts_test1": "true",