Updated to use PostgresDSN.build.

This required the addtion of INST_DB_SCHEME, which I've updated in the .env files, but will
write stories to update the AWS secrets.

Removed the INST_DB_CONN from the env. If we want to keep this, I can add an if in the
field_validator to check for existence, but that makes encoding the pwd tricky again.  Unless
we assume if the INST_DB_CONN exists, that it's local dev and ignore pwd encoding.

pyproject.toml

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

src/.env.local

@@ -13,7 +13,7 @@ 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}
+INST_DB_SCHEME=postgresql+asyncpg
 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=postgresql+asyncpg

src/config.py

@@ -1,8 +1,6 @@
 import os
-import logging
-import re
 from urllib import parse
-from typing import Dict, Any
+from typing import Dict, Any, Optional
 
 from pydantic import TypeAdapter, field_validator, ValidationInfo
 from pydantic.networks import HttpUrl, PostgresDsn
@@ -17,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
+    inst_conn: Optional[PostgresDsn] = None
     auth_client: str
     auth_url: HttpUrl
     token_url: HttpUrl
@@ -36,15 +39,15 @@ def __init__(self, **data):
 
     @field_validator("inst_conn", mode="before")
     @classmethod
-    def encode_db_password(cls, postgres_dsn, info: ValidationInfo) -> Any:
-        log = logging.getLogger()
-        pwd = re.search(".*:.*:(.*)@", postgres_dsn)
-        if pwd:
-            pwd_str = pwd.group(1)
-            encoded_password = parse.quote(pwd_str, safe="")
-            return postgres_dsn.replace(pwd_str, encoded_password)
-        else:
-            log.error(f"Postgres DSN did not contain a properly formatted URL: {postgres_dsn}")
+    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:
         """

tests/app/test_config.py

@@ -2,10 +2,16 @@
 from config import Settings
 
 
-def test_psql_password_encoding():
-    mock_config = {"inst_conn": "postgresql+asyncpg://test:\\z9-/tgb76@test/test"}
+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://test:%5Cz9-%2Ftgb76@test/test"
+    assert str(settings.inst_conn) == "postgresql+asyncpg://user:%5Cz9-%2Ftgb76%23%40@test:5432/test"
 
 
 def test_jwt_opts_valid_values():