Settings

app/lib/settings.py

@@ -0,0 +1,110 @@
+"""All configuration via environment.
+
+Take note of the environment variable prefixes required for each
+settings class, except `AppSettings`.
+"""
+from typing import Literal, Optional, Union
+
+__all__ = [
+    "APISettings",
+    "AppSettings",
+    "DatabaseSettings",
+    "EmailSettings",
+    "OpenAPISettings",
+    "ServerSettings",
+]
+
+from pydantic import Extra
+from pydantic_settings import BaseSettings
+
+
+class BaseEnvSettings(BaseSettings):
+    class Config:
+        env_file = ".env"
+        env_file_encoding = "utf-8"
+        extra = Extra.ignore
+
+
+class AppSettings(BaseEnvSettings):
+    class Config:
+        case_sensitive = True
+
+    BUILD_NUMBER: str = "0"
+    DEBUG: bool = False
+    ENVIRONMENT: str = "local"
+    LOG_LEVEL: str = "INFO"
+    NAME: str = "addressbook"
+
+    @property
+    def slug(self) -> str:
+        return "-".join(s.lower() for s in self.NAME.split())
+
+
+class APISettings(BaseEnvSettings):
+    class Config:
+        env_prefix = "API_"
+        case_sensitive = True
+
+    CACHE_EXPIRATION: int = 60
+    DB_SESSION_DEPENDENCY_KEY: str = "db_session"
+    DEFAULT_PAGINATION_LIMIT: int = 100
+    DEFAULT_USER_NAME: str = "__default_user__"
+    HEALTH_PATH: str = "/health"
+    SECRET_KEY: str = "abc123"
+    USER_DEPENDENCY_KEY: str = "user"
+
+
+class OpenAPISettings(BaseEnvSettings):
+    class Config:
+        env_prefix = "OPENAPI_"
+        case_sensitive = True
+
+    TITLE: Optional[str] = "My Litestar App"
+    VERSION: str = "0.1.0"
+    CONTACT_NAME: str = "My Name"
+    CONTACT_EMAIL: str = "some_human@some_domain.com"
+
+
+class DatabaseSettings(BaseEnvSettings):
+    class Config:
+        env_prefix = "DB_"
+        case_sensitive = True
+
+    ECHO: bool = False
+    ECHO_POOL: Union[bool, Literal["debug"]] = False
+    POOL_DISABLE: bool = False
+    POOL_MAX_OVERFLOW: int = 10
+    POOL_SIZE: int = 5
+    POOL_TIMEOUT: int = 30
+    HOST: str = "localhost"
+    PORT: int = 5432
+    NAME: str = "db-name"
+    USER: str = "db-user"
+    PASSWORD: str = "db-password"
+
+
+class ServerSettings(BaseEnvSettings):
+    class Config:
+        env_prefix = "UVICORN_"
+        case_sensitive = True
+
+    HOST: str = "localhost"
+    LOG_LEVEL: str = "info"
+    PORT: int = 8000
+    RELOAD: bool = True
+    KEEPALIVE: int = 65
+
+
+class EmailSettings(BaseEnvSettings):
+    class Config:
+        env_prefix = "EMAIL_"
+        case_sensitive = True
+
+
+# `.parse_obj()` thing is a workaround for pyright and pydantic interplay, see:
+# https://github.com/pydantic/pydantic/issues/3753#issuecomment-1087417884
+api = APISettings.parse_obj({})
+app = AppSettings.parse_obj({})
+db = DatabaseSettings.parse_obj({})
+openapi = OpenAPISettings.parse_obj({})
+server = ServerSettings.parse_obj({})

app/lib/sqlalchemy_plugin.py

@@ -1,10 +1,10 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
-from typing import TYPE_CHECKING, cast, Literal
+from typing import TYPE_CHECKING, cast
 from uuid import UUID
 
 import msgspec
+import sqlalchemy
 from litestar.contrib.sqlalchemy.plugins.init import SQLAlchemyInitPlugin
 from litestar.contrib.sqlalchemy.plugins.init.config import SQLAlchemyAsyncConfig
 from litestar.contrib.sqlalchemy.plugins.init.config.common import (
@@ -16,28 +16,7 @@ from sqlalchemy import event
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 from sqlalchemy.pool import NullPool
 
-
-DB_HOST = "localhost"
-DB_PORT = 5432
-DB_NAME = "addressbook"
-DB_USER = "addressbook"
-DB_PASSWORD = "addressbook"
-
-
-@dataclass
-class DatabaseSettings:
-    URL: str = f"postgresql+asyncpg://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
-    ECHO: bool = True
-    ECHO_POOL: bool | Literal["debug"] = False
-    POOL_DISABLE: bool = False
-    POOL_MAX_OVERFLOW: int = 10
-    POOL_SIZE: int = 5
-    POOL_TIMEOUT: int = 30
-    DB_SESSION_DEPENDENCY_KEY: str = "db_session"
-
-
-settings = DatabaseSettings()
-
+from app.lib import settings
 
 if TYPE_CHECKING:
     from typing import Any
@@ -58,15 +37,24 @@ def _default(val: Any) -> str:
     raise TypeError()
 
 
+db_connection_url = sqlalchemy.engine.URL.create(
+    drivername="postgresql+asyncpg",
+    username=settings.db.USER,
+    password=settings.db.PASSWORD,
+    host=settings.db.HOST,
+    port=settings.db.PORT,
+    database=settings.db.NAME,
+)
+
 engine = create_async_engine(
-    settings.URL,
-    echo=settings.ECHO,
-    echo_pool=settings.ECHO_POOL,
+    db_connection_url,
+    echo=settings.db.ECHO,
+    echo_pool=settings.db.ECHO_POOL,
     json_serializer=msgspec.json.Encoder(enc_hook=_default),
-    max_overflow=settings.POOL_MAX_OVERFLOW,
-    pool_size=settings.POOL_SIZE,
-    pool_timeout=settings.POOL_TIMEOUT,
-    poolclass=NullPool if settings.POOL_DISABLE else None,
+    max_overflow=settings.db.POOL_MAX_OVERFLOW,
+    pool_size=settings.db.POOL_SIZE,
+    pool_timeout=settings.db.POOL_TIMEOUT,
+    poolclass=NullPool if settings.db.POOL_DISABLE else None,
 )
 """Configure via DatabaseSettings.
 
@@ -74,7 +62,9 @@ Overrides default JSON
 serializer to use `msgspec`. See [`create_async_engine()`][sqlalchemy.ext.asyncio.create_async_engine]
 for detailed instructions.
 """
-async_session_factory = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
+async_session_factory = async_sessionmaker(
+    engine, expire_on_commit=False, class_=AsyncSession
+)
 """Database session factory.
 
 See [`async_sessionmaker()`][sqlalchemy.ext.asyncio.async_sessionmaker].
@@ -124,10 +114,11 @@ async def before_send_handler(message: Message, scope: Scope) -> None:
 
     Args:
         message: ASGI message
-        _:
         scope: ASGI scope
     """
-    session = cast("AsyncSession | None", get_litestar_scope_state(scope, SESSION_SCOPE_KEY))
+    session = cast(
+        "AsyncSession | None", get_litestar_scope_state(scope, SESSION_SCOPE_KEY)
+    )
     try:
         if session is not None and message["type"] == "http.response.start":
             if 200 <= message["status"] < 300:
@@ -141,7 +132,7 @@ async def before_send_handler(message: Message, scope: Scope) -> None:
 
 
 config = SQLAlchemyAsyncConfig(
-    session_dependency_key=settings.DB_SESSION_DEPENDENCY_KEY,
+    session_dependency_key=settings.api.DB_SESSION_DEPENDENCY_KEY,
     engine_instance=engine,
     session_maker=async_session_factory,
     before_send_handler=before_send_handler,