Refactor Settings creation

Author: DiamondJoseph

tiled/authn_database/connection_pool.py

@@ -1,7 +1,7 @@
 from fastapi import Depends
 from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
 
-from ..server.settings import get_settings
+from ..server.settings import Settings, get_settings
 from ..utils import ensure_specified_sql_driver
 
 # A given process probably only has one of these at a time, but we
@@ -31,9 +31,9 @@ async def close_database_connection_pool(database_settings):
         await engine.dispose()
 
 
-async def get_database_engine(settings=Depends(get_settings)):
+async def get_database_engine(settings: Settings = Depends(get_settings)):
     # Special case for single-user mode
-    if settings.database_uri is None:
+    if settings.database_settings.uri is None:
         return None
     try:
         return _connection_pools[settings.database_settings]

tiled/server/app.py

@@ -51,7 +51,7 @@
 from .compression import CompressionMiddleware
 from .dependencies import get_root_tree
 from .router import get_router
-from .settings import get_settings
+from .settings import Settings, get_settings
 from .utils import (
     API_KEY_COOKIE_NAME,
     CSRF_COOKIE_NAME,
@@ -461,29 +461,31 @@ def override_get_settings():
             if server_settings.get(item) is not None:
                 setattr(settings, item, server_settings[item])
         database = server_settings.get("database", {})
-        if database.get("uri"):
-            settings.database_uri = database["uri"]
-        if database.get("pool_size"):
-            settings.database_pool_size = database["pool_size"]
-        if database.get("pool_pre_ping"):
-            settings.database_pool_pre_ping = database["pool_pre_ping"]
-        if database.get("max_overflow"):
-            settings.database_max_overflow = database["max_overflow"]
-        if database.get("init_if_not_exists"):
-            settings.database_init_if_not_exists = database["init_if_not_exists"]
+        if uri := database.get("uri"):
+            settings.database_settings.uri = uri
+        if pool_size := database.get("pool_size"):
+            settings.database_settings.pool_size = pool_size
+        if pool_pre_ping := database.get("pool_pre_ping"):
+            settings.database_settings.pool_pre_ping = pool_pre_ping
+        if max_overflow := database.get("max_overflow"):
+            settings.database_settings.max_overflow = max_overflow
+        if init_if_not_exists := database.get("init_if_not_exists"):
+            settings.database_init_if_not_exists = init_if_not_exists
         if authentication.get("providers"):
             # If we support authentication providers, we need a database, so if one is
             # not set, use a SQLite database in memory. Horizontally scaled deployments
             # must specify a persistent database.
-            settings.database_uri = settings.database_uri or "sqlite://"
+            settings.database_settings.uri = (
+                settings.database_settings.uri or "sqlite://"
+            )
         return settings
 
     async def startup_event():
         from .. import __version__
 
         logger.info(f"Tiled version {__version__}")
         # Validate the single-user API key.
-        settings = app.dependency_overrides[get_settings]()
+        settings: Settings = app.dependency_overrides[get_settings]()
         single_user_api_key = settings.single_user_api_key
         API_KEY_MSG = """
 Here are two ways to generate a good API key:
@@ -535,7 +537,7 @@ async def startup_event():
         # client.context.app.state.root_tree
         app.state.root_tree = app.dependency_overrides[get_root_tree]()
 
-        if settings.database_uri is not None:
+        if settings.database_settings.uri is not None:
             from sqlalchemy.ext.asyncio import AsyncSession
 
             from ..alembic_utils import (
@@ -660,8 +662,8 @@ async def shutdown_event():
         for task in tasks.get("shutdown", []):
             await task()
 
-        settings = app.dependency_overrides[get_settings]()
-        if settings.database_uri is not None:
+        settings: Settings = app.dependency_overrides[get_settings]()
+        if settings.database_settings.uri is not None:
             from ..authn_database.connection_pool import close_database_connection_pool
 
             for task in app.state.tasks:

tiled/server/settings.py

@@ -1,82 +1,50 @@
-import collections
 import os
 import secrets
 from datetime import timedelta
 from functools import cache
 from typing import Any, List, Optional
 
-from pydantic_settings import BaseSettings
+from pydantic import Field
+from pydantic.dataclasses import dataclass
+from pydantic_settings import BaseSettings, SettingsConfigDict
 
-DatabaseSettings = collections.namedtuple(
-    "DatabaseSettings", "uri pool_size pool_pre_ping max_overflow"
-)
+
+# hashable cache key for use in tiled.authn_database.connection_pool
+@dataclass(unsafe_hash=True)
+class DatabaseSettings:
+    uri: Optional[str] = None
+    pool_size: int = 5
+    pool_pre_ping: bool = True
+    max_overflow: int = 5
 
 
 class Settings(BaseSettings):
     tree: Any = None
-    allow_anonymous_access: bool = bool(
-        int(os.getenv("TILED_ALLOW_ANONYMOUS_ACCESS", False))
-    )
-    allow_origins: List[str] = [
-        item for item in os.getenv("TILED_ALLOW_ORIGINS", "").split() if item
-    ]
+    allow_anonymous_access: bool = False
+    allow_origins: List[str] = Field(default_factory=list)
     authenticator: Any = None
     # These 'single user' settings are only applicable if authenticator is None.
-    single_user_api_key: str = os.getenv(
-        "TILED_SINGLE_USER_API_KEY", secrets.token_hex(32)
-    )
-    single_user_api_key_generated: bool = not (
-        "TILED_SINGLE_USER_API_KEY" in os.environ
-    )
+    single_user_api_key: str = secrets.token_hex(32)
+    single_user_api_key_generated: bool = "TILED_SINGLE_USER_API_KEY" not in os.environ
     # The TILED_SERVER_SECRET_KEYS may be a single key or a ;-separated list of
     # keys to support key rotation. The first key will be used for encryption. Each
     # key will be tried in turn for decryption.
-    secret_keys: List[str] = os.getenv(
-        "TILED_SERVER_SECRET_KEYS", secrets.token_hex(32)
-    ).split(";")
-    access_token_max_age: timedelta = timedelta(
-        seconds=int(os.getenv("TILED_ACCESS_TOKEN_MAX_AGE", 15 * 60))  # 15 minutes
-    )
-    refresh_token_max_age: timedelta = timedelta(
-        seconds=int(
-            os.getenv("TILED_REFRESH_TOKEN_MAX_AGE", 7 * 24 * 60 * 60)
-        )  # 7 days
-    )
-    session_max_age: Optional[timedelta] = timedelta(
-        seconds=int(os.getenv("TILED_SESSION_MAX_AGE", 365 * 24 * 60 * 60))  # 365 days
-    )
+    secret_keys: List[str] = [secrets.token_hex(32)]
+    access_token_max_age: timedelta = 15 * 60  # 15 minutes
+    refresh_token_max_age: timedelta = 7 * 24 * 60 * 60  # 7 days
+    session_max_age: timedelta = 365 * 24 * 60 * 60  # 365 days
     # Put a fairly low limit on the maximum size of one chunk, keeping in mind
     # that data should generally be chunked. When we implement async responses,
     # we can raise this global limit.
-    response_bytesize_limit: int = int(
-        os.getenv("TILED_RESPONSE_BYTESIZE_LIMIT", 300_000_000)
-    )  # 300 MB
-    reject_undeclared_specs: bool = bool(
-        int(os.getenv("TILED_REJECT_UNDECLARED_SPECS", 0))
-    )
-    database_uri: Optional[str] = os.getenv("TILED_DATABASE_URI")
-    database_init_if_not_exists: bool = int(
-        os.getenv("TILED_DATABASE_INIT_IF_NOT_EXISTS", False)
-    )
-    database_pool_size: Optional[int] = int(os.getenv("TILED_DATABASE_POOL_SIZE", 5))
-    database_pool_pre_ping: Optional[bool] = bool(
-        int(os.getenv("TILED_DATABASE_POOL_PRE_PING", 1))
-    )
-    database_max_overflow: Optional[int] = int(
-        os.getenv("TILED_DATABASE_MAX_OVERFLOW", 5)
-    )
+    response_bytesize_limit: int = 300_000_000  # 300 MB
+    reject_undeclared_specs: bool = False
+    database_settings: DatabaseSettings = Field(DatabaseSettings(), alias="database")
+    database_init_if_not_exists: bool = False
     expose_raw_assets: bool = True
 
-    @property
-    def database_settings(self):
-        # The point of this alias is to return a hashable cache key for use in
-        # the module tiled.authn_database.connection_pool.
-        return DatabaseSettings(
-            uri=self.database_uri,
-            pool_size=self.database_pool_size,
-            pool_pre_ping=self.database_pool_pre_ping,
-            max_overflow=self.database_max_overflow,
-        )
+    model_config = SettingsConfigDict(
+        env_prefix="TILED_", nested_model_default_partial_update=True
+    )
 
 
 @cache