refactor(conf): support env alongside file configuration

Author: matthiasschaub

config/test.config.toml

@@ -1 +1,2 @@
 # Empty config file for tests
+user_agent = "foo"  # Will be overwritten by pytest.env

sketch_map_tool/__init__.py

@@ -9,23 +9,23 @@
 # https://github.com/GIScience/sketch-map-tool/issues/503
 from osgeo import gdal, osr  # noqa: F401
 
-from sketch_map_tool.config import get_config_value
+from sketch_map_tool.config import CONFIG
 from sketch_map_tool.database import client_flask as db_client
 from sketch_map_tool.definitions import LANGUAGES
 
 __version__ = "2025.11.12"
 
 # Setup logging
-LEVEL = getattr(logging, get_config_value("log-level").upper())
+LEVEL = getattr(logging, CONFIG.log_level.upper())
 FORMAT = "%(asctime)s - %(levelname)s - %(filename)s - %(funcName)s - %(message)s"
 logging.basicConfig(
     level=LEVEL,
     format=FORMAT,
 )
 
 CELERY_CONFIG = {
-    "broker_url": get_config_value("broker-url"),
-    "result_backend": get_config_value("result-backend"),
+    "broker_url": CONFIG.broker_url,
+    "result_backend": CONFIG.result_backend,
     "task_serializer": "pickle",
     "task_track_started": True,  # report ‘started’ status worker executes task
     "task_send_sent_event": True,

sketch_map_tool/config.py

@@ -1,35 +1,16 @@
+import logging
 import os
-from types import MappingProxyType
-from typing import Dict
 
-import toml
+from pydantic import field_validator
+from pydantic_settings import (
+    BaseSettings,
+    PydanticBaseSettingsSource,
+    SettingsConfigDict,
+    TomlConfigSettingsSource,
+)
 
 from sketch_map_tool.helpers import get_project_root
 
-DEFAULT_CONFIG = {
-    "data-dir": str(get_project_root() / "data"),
-    "weights-dir": str(get_project_root() / "weights"),
-    "user-agent": "sketch-map-tool",
-    "broker-url": "redis://localhost:6379",
-    "result-backend": "db+postgresql://smt:smt@localhost:5432",
-    "cleanup-map-frames-interval": "12 months",
-    "wms-url-osm": "https://maps.heigit.org/raster/osm-carto/service?SERVICE=WMS&VERSION=1.1.1",
-    "wms-layers-osm": "heigit:osm-carto-proxy",
-    "wms-url-esri-world-imagery": "https://maps.heigit.org/raster/sketch-map-tool/service?SERVICE=WMS&VERSION=1.1.1",
-    "wms-url-esri-world-imagery-fallback": "https://maps.heigit.org/raster/sketch-map-tool/service?SERVICE=WMS&VERSION=1.1.1",
-    "wms-layers-esri-world-imagery": "world_imagery",
-    "wms-layers-esri-world-imagery-fallback": "world_imagery_fallback",
-    "wms-read-timeout": 600,
-    "max-nr-simultaneous-uploads": 100,
-    "yolo_cls": "SMT-CLS",
-    "yolo_osm_obj": "SMT-OSM",
-    "yolo_esri_obj": "SMT-ESRI",
-    "model_type_sam": "vit_b",
-    "esri-api-key": "",
-    "log-level": "INFO",
-    "point-area-threshold": 0.00047,
-}
-
 
 def get_config_path() -> str:
     """Get configuration file path.
@@ -41,26 +22,55 @@ def get_config_path() -> str:
     return os.getenv("SMT_CONFIG", default=default)
 
 
-def load_config_from_file(path: str) -> Dict[str, str]:
-    """Load configuration from file on disk."""
-    if os.path.isfile(path):
-        with open(path, "r") as f:
-            return toml.load(f)
-    else:
-        return {}
+class Config(BaseSettings):
+    broker_url: str = "redis://localhost:6379"
+    cleanup_map_frames_interval: str = "12 months"
+    data_dir: str = str(get_project_root() / "data")  # TODO: make this a Path
+    esri_api_key: str = ""
+    log_level: str = "INFO"
+    max_nr_simultaneous_uploads: int = 100
+    model_type_sam: str = "vit_b"
+    point_area_threshold: float = 0.00047
+    result_backend: str = "db+postgresql://smt:smt@localhost:5432"
+    user_agent: str = "sketch-map-tool"
+    weights_dir: str = str(get_project_root() / "weights")  # TODO: make this a Path
+    wms_layers_esri_world_imagery: str = "world_imagery"
+    wms_layers_esri_world_imagery_fallback: str = "world_imagery_fallback"
+    wms_layers_osm: str = "heigit:osm-carto-proxy"
+    wms_read_timeout: int = 600
+    wms_url_esri_world_imagery: str = "https://maps.heigit.org/raster/sketch-map-tool/service?SERVICE=WMS&VERSION=1.1.1"
+    wms_url_esri_world_imagery_fallback: str = "https://maps.heigit.org/raster/sketch-map-tool/service?SERVICE=WMS&VERSION=1.1.1"
+    wms_url_osm: str = (
+        "https://maps.heigit.org/raster/osm-carto/service?SERVICE=WMS&VERSION=1.1.1"
+    )
+    yolo_cls: str = "SMT-CLS"
+    yolo_esri_obj: str = "SMT-ESRI"
+    yolo_osm_obj: str = "SMT-OSM"
 
+    model_config = SettingsConfigDict(
+        env_prefix="SMT_",
+        toml_file=get_config_path(),
+    )
 
-def get_config() -> MappingProxyType:
-    """Get configuration variables from environment and file.
+    @classmethod
+    def settings_customise_sources(
+        cls,
+        settings_cls: type[BaseSettings],
+        init_settings: PydanticBaseSettingsSource,
+        env_settings: PydanticBaseSettingsSource,
+        dotenv_settings: PydanticBaseSettingsSource,
+        file_secret_settings: PydanticBaseSettingsSource,
+    ) -> tuple[PydanticBaseSettingsSource, ...]:
+        # https://docs.pydantic.dev/latest/concepts/pydantic_settings/#other-settings-source
+        # env takes precedence over file settings
+        return (env_settings, TomlConfigSettingsSource(settings_cls))
 
-    Configuration values from file will be given precedence over default values.
-    """
-    cfg = DEFAULT_CONFIG
-    cfg_file = load_config_from_file(get_config_path())
-    cfg.update(cfg_file)
-    return MappingProxyType(cfg)
+    @field_validator("esri_api_key", mode="before")
+    @classmethod
+    def check_esri_api_key(cls, value: str) -> str:
+        if not value:
+            logging.warning("No ESRI API Key found.")
+        return value
 
 
-def get_config_value(key: str):
-    config = get_config()
-    return config[key]
+CONFIG = Config()

sketch_map_tool/database/client_celery.py

@@ -7,7 +7,7 @@
 from psycopg2.extensions import connection
 
 from sketch_map_tool import __version__
-from sketch_map_tool.config import get_config_value
+from sketch_map_tool.config import CONFIG
 from sketch_map_tool.exceptions import (
     CustomFileDoesNotExistAnymoreError,
     CustomFileNotFoundError,
@@ -20,7 +20,7 @@
 
 def open_connection():
     global db_conn
-    raw = get_config_value("result-backend")
+    raw = CONFIG.result_backend
     dns = raw[3:]
     db_conn = psycopg2.connect(dns)
     db_conn.autocommit = True
@@ -129,7 +129,7 @@ def cleanup_map_frames():
     """
     with db_conn.cursor() as curs:
         try:
-            curs.execute(query, [get_config_value("cleanup-map-frames-interval")])
+            curs.execute(query, [CONFIG.cleanup_map_frames_interval])
         except UndefinedTable:
             logging.info("Table `map_frame` does not exist yet. Nothing todo.")
 

sketch_map_tool/database/client_flask.py

@@ -6,7 +6,7 @@
 from psycopg2.extensions import connection
 from werkzeug.utils import secure_filename
 
-from sketch_map_tool.config import get_config_value
+from sketch_map_tool.config import CONFIG
 from sketch_map_tool.exceptions import (
     CustomFileDoesNotExistAnymoreError,
     CustomFileNotFoundError,
@@ -18,7 +18,7 @@
 
 def open_connection():
     if "db_conn" not in g:
-        raw = get_config_value("result-backend")
+        raw = CONFIG.result_backend
         dns = raw[3:]
         g.db_conn = psycopg2.connect(dns)
         g.db_conn.autocommit = True

sketch_map_tool/definitions.py

@@ -7,7 +7,7 @@
 import requests
 from werkzeug.utils import secure_filename
 
-from sketch_map_tool.config import get_config_value
+from sketch_map_tool.config import CONFIG
 from sketch_map_tool.models import LiteratureReference, PaperFormat
 from sketch_map_tool.openaerialmap import client as oam_client
 
@@ -37,7 +37,7 @@ def get_attribution(layer: str) -> str:
         url = (
             "https://basemaps-api.arcgis.com/arcgis/rest/services/styles/ArcGIS:Imagery"
         )
-        token = get_config_value("esri-api-key")
+        token = CONFIG.esri_api_key
         if token == "":
             sources = "Esri, Maxar, Earthstar Geographics, and the GIS User Community"
             logging.warning(
@@ -69,7 +69,7 @@ def get_literature_references() -> list[LiteratureReference]:
     For image source either a web URL or a filename of a file in the publications folder
     is expected.
     """
-    p = Path(get_config_value("data-dir")) / "literature.json"
+    p = Path(CONFIG.data_dir) / "literature.json"
     with open(p, "r") as f:
         raw = json.load(f)
 

sketch_map_tool/routes.py

@@ -81,7 +81,7 @@ def create(lang="en") -> str:
     return render_template(
         "create.html.jinja",
         lang=lang,
-        esri_api_key=config.get_config_value("esri-api-key"),
+        esri_api_key=config.CONFIG.esri_api_key,
     )
 
 

sketch_map_tool/tasks.py

@@ -10,8 +10,8 @@
 from ultralytics import YOLO
 from ultralytics_MB import YOLO as YOLO_MB
 
+from sketch_map_tool import CONFIG, map_generation
 from sketch_map_tool import celery_app as celery
-from sketch_map_tool import get_config_value, map_generation
 from sketch_map_tool.database import client_celery as db_client_celery
 from sketch_map_tool.definitions import get_attribution
 from sketch_map_tool.exceptions import MarkingDetectionError
@@ -64,9 +64,9 @@ def init_worker_ml_models(**_):
     )
     sam_predictor = SAM2ImagePredictor(sam2_model)
 
-    yolo_obj_osm = YOLO_MB(init_model(get_config_value("yolo_osm_obj")))
-    yolo_obj_esri = YOLO_MB(init_model(get_config_value("yolo_esri_obj")))
-    yolo_cls = YOLO(init_model(get_config_value("yolo_cls")))
+    yolo_obj_osm = YOLO_MB(init_model(CONFIG.yolo_osm_obj))
+    yolo_obj_esri = YOLO_MB(init_model(CONFIG.yolo_esri_obj))
+    yolo_cls = YOLO(init_model(CONFIG.yolo_cls))
 
 
 @worker_process_shutdown.connect
@@ -78,7 +78,7 @@ def shutdown_worker(**_):
 
 @setup_logging.connect
 def on_setup_logging(**_):
-    level = getattr(logging, get_config_value("log-level").upper())
+    level = getattr(logging, CONFIG.log_level.upper())
     format = "%(asctime)s - %(levelname)s - %(filename)s - %(funcName)s - %(message)s"
     logging.basicConfig(
         level=level,

sketch_map_tool/upload_processing/ml_models.py

@@ -5,20 +5,20 @@
 import torch
 from torch._prims_common import DeviceLikeType
 
-from sketch_map_tool.config import get_config_value
+from sketch_map_tool.config import CONFIG
 
 
 def init_model(id: str) -> Path:
     """Initialize model. Raise error if not found."""
-    raw = Path(get_config_value("weights-dir")) / id
+    raw = Path(CONFIG.weights_dir) / id
     path = raw.with_suffix(".pt")
     if not path.is_file():
         raise FileNotFoundError("Model not found at " + str(path))
     return path
 
 
 def init_sam2(id: str = "sam2_hiera_base_plus") -> Path:
-    raw = Path(get_config_value("weights-dir")) / id
+    raw = Path(CONFIG.weights_dir) / id
     path = raw.with_suffix(".pt")
     base_url = "https://dl.fbaipublicfiles.com/segment_anything_2/072824/"
     url = base_url + id + ".pt"

sketch_map_tool/upload_processing/post_process.py

@@ -8,7 +8,7 @@
 from shapely.ops import transform, unary_union
 from shapelysmooth import chaikin_smooth
 
-from sketch_map_tool.config import get_config_value
+from sketch_map_tool.config import CONFIG
 from sketch_map_tool.definitions import COLORS
 from sketch_map_tool.models import Bbox
 
@@ -157,7 +157,7 @@ def smooth(fc: FeatureCollection) -> FeatureCollection:
 
 def classify_points(fc: FeatureCollection, bbox: Bbox) -> FeatureCollection:
     """Classify each feature as point or polygon based area."""
-    point_area_threshold = get_config_value("point-area-threshold")
+    point_area_threshold = CONFIG.point_area_threshold
     fc_ = FeatureCollection(features=[])
 
     for feature in fc["features"]: