Use pydantic settings management

alembic/env.py

@@ -1,13 +1,13 @@
 """Alembic environment configuration."""
 
 from logging.config import fileConfig
-from os import getenv as os_getenv
 
 from sqlalchemy import create_engine
 from sqlalchemy.pool import NullPool
 
 from alembic import context
 from dds_glossary.model import Base
+from dds_glossary.settings import get_settings
 
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.
@@ -20,9 +20,9 @@
 
 # add your model's MetaData object here
 # for 'autogenerate' support
+database_url = get_settings().DATABASE_URL.get_secret_value()
 target_metadata = Base.metadata
-from dds_glossary.database import DATABASE_URL
-config.set_main_option("sqlalchemy.url", DATABASE_URL)
+config.set_main_option("sqlalchemy.url", database_url)
 
 
 # other values from the config, defined by the needs of env.py,

dds_glossary/__init__.py

@@ -1,9 +1,5 @@
 """dds_glossary package."""
 
-from dotenv import load_dotenv
-
 __version__ = "0.1.0"
 
 __all__ = ["__version__"]
-
-load_dotenv()

dds_glossary/auth.py

@@ -1,11 +1,12 @@
 """Authentication utils for the dds_glossary package."""
 
 from http import HTTPStatus
-from os import getenv as os_getenv
 
 from fastapi import Depends, HTTPException
 from fastapi.security import APIKeyHeader
 
+from .settings import get_settings
+
 api_key_header = APIKeyHeader(name="X-API-Key")
 
 
@@ -23,7 +24,7 @@ def get_api_key(api_key: str = Depends(api_key_header)) -> dict[str, str]:
         HTTPException: If the API key is missing or invalid. Or if the API key
             environment variable is missing.
     """
-    correct_api_key: str | None = os_getenv("API_KEY")
+    correct_api_key: str | None = get_settings().API_KEY.get_secret_value()
     if correct_api_key is None or api_key != correct_api_key:
         raise HTTPException(
             status_code=HTTPStatus.FORBIDDEN,

dds_glossary/main.py

@@ -1,14 +1,14 @@
 """Main entry for the dds_glossary server."""
 
 import argparse
-from os import getenv as os_getenv
 
 import sentry_sdk
 import uvicorn
 from fastapi import FastAPI
 from fastapi_versioning import VersionedFastAPI
 
 from .routes import router_non_versioned, router_versioned
+from .settings import get_settings
 
 
 def create_app() -> FastAPI:
@@ -17,7 +17,7 @@ def create_app() -> FastAPI:
         FastAPI: the application object
     """
     sentry_sdk.init(
-        dsn=os_getenv("SENTRY_DSN"),
+        dsn=get_settings().SENTRY_DSN.get_secret_value(),
         # Set traces_sample_rate to 1.0 to capture 100%
         # of transactions for performance monitoring.
         traces_sample_rate=1.0,
@@ -48,5 +48,5 @@ def create_app() -> FastAPI:
     uvicorn.run(
         "dds_glossary.main:create_app",
         reload=args.f_reload,
-        host=os_getenv("HOST_IP", "127.0.0.1"),
+        host=get_settings().HOST_IP,
     )

dds_glossary/settings.py

@@ -0,0 +1,22 @@
+"""Settings for dds_glossary package."""
+
+from functools import lru_cache
+
+from pydantic import SecretStr
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    """Settings for dds_glossary package."""
+
+    API_KEY: SecretStr
+    DATABASE_URL: SecretStr
+    SENTRY_DSN: SecretStr = SecretStr("")
+
+    HOST_IP: str = "127.0.0.1"
+
+
+@lru_cache()
+def get_settings():
+    """Returns settings object."""
+    return Settings()

pyproject.toml

@@ -34,7 +34,7 @@ dependencies = [
     "lxml",
     "owlready2",
     "psycopg",
-    "python-dotenv",
+    "pydantic_settings~=2.0",
     "sentry-sdk[fastapi]",
     "SQLAlchemy>=2.0.0",
     "sqlalchemy-utils",

tests/unit/test_auth.py

@@ -7,6 +7,7 @@
 from pytest import raises as pytest_raises
 
 from dds_glossary.auth import get_api_key
+from dds_glossary.settings import get_settings
 
 
 def _validate_error(exc_info: ExceptionInfo) -> None:
@@ -36,8 +37,7 @@ def test_get_api_key_invalid_key() -> None:
     _validate_error(exc_info)
 
 
-def test_get_api_key_valid_key(monkeypatch: MonkeyPatch) -> None:
+def test_get_api_key_valid_key() -> None:
     """Test the get_api_key function with a valid key."""
-    api_key = "valid"
-    monkeypatch.setenv("API_KEY", api_key)
-    assert get_api_key(api_key="valid") == {"api_key": api_key}
+    api_key = get_settings().API_KEY.get_secret_value()
+    assert get_api_key(api_key=api_key) == {"api_key": api_key}

tests/unit/test_routes.py

@@ -7,6 +7,7 @@
 
 from dds_glossary.model import Dataset, FailedDataset
 from dds_glossary.schema import InitDatasetsResponse, VersionResponse
+from dds_glossary.settings import get_settings
 
 
 def test_version(
@@ -53,9 +54,8 @@ def test_init_datasets_valid_key(client: TestClient, monkeypatch: MonkeyPatch) -
             failed_datasets=failed_datasets,
         ),
     )
-    api_key = "valid"
-    monkeypatch.setenv("API_KEY", api_key)
 
+    api_key = get_settings().API_KEY.get_secret_value()
     response = client.post("/latest/init_datasets", headers={"X-API-Key": api_key})
     assert response.status_code == HTTPStatus.OK
     assert response.headers["content-type"] == "application/json"