feat: use pydantic-settings as default config format

apps/api/src/auth.py

@@ -3,9 +3,9 @@
 from fastapi import HTTPException, Security, status
 from fastapi.security.api_key import APIKeyQuery
 
-from src.config import API_KEY_NAME, API_KEY
+from src.config import settings
 
-api_key_query = APIKeyQuery(name=API_KEY_NAME, auto_error=False)
+api_key_query = APIKeyQuery(name=settings.API_KEY_NAME, auto_error=False)
 
 
 def verify_key(input_key: str, true_key: str):
@@ -16,4 +16,4 @@ def verify_key(input_key: str, true_key: str):
 
 
 def verify_api_key(api_key_query: str = Security(api_key_query)):
-    verify_key(api_key_query, API_KEY)
+    verify_key(api_key_query, settings.API_KEY)

apps/api/src/config.py

@@ -2,65 +2,82 @@
 import tempfile
 import gettext
 import secrets
+from typing import Optional
 
-from fastfetchbot_shared.utils.parse import get_env_bool
-
-env = os.environ
-current_directory = os.path.dirname(os.path.abspath(__file__))
-conf_dir = os.path.join(current_directory, "..", "conf")
-
-# FastAPI environment variables
-BASE_URL = env.get("BASE_URL", "localhost")
-API_KEY_NAME = env.get("API_KEY_NAME", "pwd")
-API_KEY = env.get("API_KEY", secrets.token_urlsafe(32))
-
-# Filesystem environment variables
-TEMP_DIR = env.get("TEMP_DIR", tempfile.gettempdir())
-WORK_DIR = env.get("WORK_DIR", os.getcwd())
-DOWNLOAD_DIR = env.get("DOWNLOAD_DIR", os.path.join(WORK_DIR, "download"))
-DEBUG_MODE = get_env_bool(env, "DEBUG_MODE", False)
-
-# Logging environment variables
-LOG_FILE_PATH = env.get("LOG_FILE_PATH", TEMP_DIR)
-LOG_LEVEL = env.get("LOG_LEVEL", "DEBUG")
-
-# MongoDB environment variables
-DATABASE_ON = get_env_bool(env, "DATABASE_ON", False)
-MONGODB_PORT = int(env.get("MONGODB_PORT", 27017)) or 27017
-MONGODB_HOST = env.get("MONGODB_HOST", "localhost")
-MONGODB_URL = env.get("MONGODB_URL", f"mongodb://{MONGODB_HOST}:{MONGODB_PORT}")
-
-# File exporter toggle (used by telegram bot to show/hide buttons)
-FILE_EXPORTER_ON = get_env_bool(env, "FILE_EXPORTER_ON", True)
-DOWNLOAD_VIDEO_TIMEOUT = env.get("DOWNLOAD_VIDEO_TIMEOUT", 600)
-
-# Celery configuration
-CELERY_BROKER_URL = env.get("CELERY_BROKER_URL", "redis://localhost:6379/0")
-CELERY_RESULT_BACKEND = env.get("CELERY_RESULT_BACKEND", "redis://localhost:6379/1")
-
-# AWS storage
-AWS_STORAGE_ON = get_env_bool(env, "AWS_STORAGE_ON", False)
-AWS_ACCESS_KEY_ID = env.get("AWS_ACCESS_KEY_ID", None)
-AWS_SECRET_ACCESS_KEY = env.get("AWS_SECRET_ACCESS_KEY", None)
-AWS_S3_BUCKET_NAME = env.get("AWS_S3_BUCKET_NAME", "")
-AWS_REGION_NAME = env.get("AWS_REGION_NAME", "")
-AWS_DOMAIN_HOST = env.get("AWS_DOMAIN_HOST", None)
-if not (AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY and AWS_S3_BUCKET_NAME):
-    AWS_STORAGE_ON = False
-
-# Inoreader
-INOREADER_APP_ID = env.get("INOREADER_APP_ID", None)
-INOREADER_APP_KEY = env.get("INOREADER_APP_KEY", None)
-INOREADER_EMAIL = env.get("INOREADER_EMAIL", None)
-INOREADER_PASSWORD = env.get("INOREADER_PASSWORD", None)
-
-# Locale directories environment variables
+from pydantic import Field, model_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class ApiSettings(BaseSettings):
+    model_config = SettingsConfigDict(extra="ignore")
+
+    # FastAPI
+    BASE_URL: str = "localhost"
+    API_KEY_NAME: str = "pwd"
+    API_KEY: str = Field(default_factory=lambda: secrets.token_urlsafe(32))
+
+    # Filesystem
+    TEMP_DIR: str = tempfile.gettempdir()
+    WORK_DIR: str = os.getcwd()
+    DOWNLOAD_DIR: str = ""
+    DEBUG_MODE: bool = False
+
+    # Logging
+    LOG_FILE_PATH: str = ""
+    LOG_LEVEL: str = "DEBUG"
+
+    # MongoDB
+    DATABASE_ON: bool = False
+    MONGODB_PORT: int = 27017
+    MONGODB_HOST: str = "localhost"
+    MONGODB_URL: str = ""
+
+    # File exporter
+    FILE_EXPORTER_ON: bool = True
+    DOWNLOAD_VIDEO_TIMEOUT: int = 600
+
+    # Celery
+    CELERY_BROKER_URL: str = "redis://localhost:6379/0"
+    CELERY_RESULT_BACKEND: str = "redis://localhost:6379/1"
+
+    # AWS storage
+    AWS_STORAGE_ON: bool = False
+    AWS_ACCESS_KEY_ID: Optional[str] = None
+    AWS_SECRET_ACCESS_KEY: Optional[str] = None
+    AWS_S3_BUCKET_NAME: str = ""
+    AWS_REGION_NAME: str = ""
+    AWS_DOMAIN_HOST: Optional[str] = None
+
+    # Inoreader
+    INOREADER_APP_ID: Optional[str] = None
+    INOREADER_APP_KEY: Optional[str] = None
+    INOREADER_EMAIL: Optional[str] = None
+    INOREADER_PASSWORD: Optional[str] = None
+
+    # Utils
+    HTTP_REQUEST_TIMEOUT: int = 30
+
+    # Telegram Bot callback URL
+    TELEGRAM_BOT_CALLBACK_URL: str = "http://telegram-bot:10451"
+
+    @model_validator(mode="after")
+    def _resolve_derived(self) -> "ApiSettings":
+        if not self.DOWNLOAD_DIR:
+            self.DOWNLOAD_DIR = os.path.join(self.WORK_DIR, "download")
+        if not self.LOG_FILE_PATH:
+            self.LOG_FILE_PATH = self.TEMP_DIR
+        if not self.MONGODB_URL:
+            self.MONGODB_URL = f"mongodb://{self.MONGODB_HOST}:{self.MONGODB_PORT}"
+        if not (self.AWS_ACCESS_KEY_ID and self.AWS_SECRET_ACCESS_KEY and self.AWS_S3_BUCKET_NAME):
+            self.AWS_STORAGE_ON = False
+        return self
+
+
+settings = ApiSettings()
+
+# --- Non-settings module-level objects ---
+
+# Locale / i18n
 localedir = os.path.join(os.path.dirname(__file__), "locale")
 translation = gettext.translation("messages", localedir=localedir, fallback=True)
 _ = translation.gettext
-
-# Utils environment variables
-HTTP_REQUEST_TIMEOUT = env.get("HTTP_REQUEST_TIMEOUT", 30)
-
-# Telegram Bot callback URL (for inter-service communication)
-TELEGRAM_BOT_CALLBACK_URL = env.get("TELEGRAM_BOT_CALLBACK_URL", "http://telegram-bot:10451")

apps/api/src/database.py

@@ -3,13 +3,13 @@
 from motor.motor_asyncio import AsyncIOMotorClient
 from beanie import init_beanie, Document, Indexed
 
-from src.config import MONGODB_URL
+from src.config import settings
 from src.models.database_model import document_list
 from fastfetchbot_shared.utils.logger import logger
 
 
 async def startup() -> None:
-    client = AsyncIOMotorClient(MONGODB_URL)
+    client = AsyncIOMotorClient(settings.MONGODB_URL)
     await init_beanie(database=client["telegram_bot"], document_models=document_list)
 
 

apps/api/src/main.py

@@ -6,7 +6,7 @@
 
 from src import database
 from src.routers import inoreader, scraper_routers, scraper
-from src.config import DATABASE_ON
+from src.config import settings
 from fastfetchbot_shared.utils.logger import logger
 
 SENTRY_DSN = ""
@@ -23,12 +23,12 @@
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    if DATABASE_ON:
+    if settings.DATABASE_ON:
         await database.startup()
     try:
         yield
     finally:
-        if DATABASE_ON:
+        if settings.DATABASE_ON:
             await database.shutdown()
 
 

apps/api/src/routers/inoreader.py

@@ -1,7 +1,7 @@
 from fastapi import APIRouter
 from fastapi.requests import Request
 
-from src.config import INOREADER_APP_ID, INOREADER_APP_KEY
+from src.config import settings
 from src.services.inoreader import Inoreader
 from src.services.inoreader.process import (
     get_inoreader_item_async,
@@ -21,7 +21,7 @@ async def get_inoreader_webhook_data(data: dict):
 
 @router.post("/triggerAsync", dependencies=[Security(verify_api_key)])
 async def inoreader_trigger_webhook(request: Request):
-    if not INOREADER_APP_ID or not INOREADER_APP_KEY:
+    if not settings.INOREADER_APP_ID or not settings.INOREADER_APP_KEY:
         return "inoreader app id or key not set"
     params = request.query_params
     await get_inoreader_item_async(trigger=True, params=params)

apps/api/src/routers/scraper.py

@@ -3,7 +3,7 @@
 from fastapi import APIRouter
 from fastapi.requests import Request
 
-from src.config import API_KEY_NAME
+from src.config import settings
 from src.services.scrapers.common import InfoExtractService
 from fastapi import Security
 from src.auth import verify_api_key
@@ -20,8 +20,8 @@ async def get_item_route(request: Request):
     url = query_params.pop("url")
     ban_list = query_params.pop("ban_list", None)
     logger.debug(f"get_item_route: url: {url}, query_params: {query_params}")
-    if API_KEY_NAME in query_params:
-        query_params.pop(API_KEY_NAME)
+    if settings.API_KEY_NAME in query_params:
+        query_params.pop(settings.API_KEY_NAME)
     url_metadata = await get_url_metadata(url, ban_list)
     item = InfoExtractService(url_metadata, **query_params)
     result = await item.get_item()

apps/api/src/services/amazon/s3.py

@@ -11,13 +11,13 @@
 
 from fastfetchbot_shared.utils.logger import logger
 from fastfetchbot_shared.utils.network import download_file_to_local
-from src.config import AWS_S3_BUCKET_NAME, AWS_REGION_NAME, AWS_DOMAIN_HOST
+from src.config import settings
 
 session = aioboto3.Session()
 image_url_host = (
-    AWS_DOMAIN_HOST
-    if AWS_DOMAIN_HOST
-    else f"{AWS_S3_BUCKET_NAME}.s3.{AWS_REGION_NAME}.amazonaws.com"
+    settings.AWS_DOMAIN_HOST
+    if settings.AWS_DOMAIN_HOST
+    else f"{settings.AWS_S3_BUCKET_NAME}.s3.{settings.AWS_REGION_NAME}.amazonaws.com"
 )
 
 
@@ -40,11 +40,13 @@ async def download_and_upload(url: str, referer: str = None, suite: str = "test"
 
 async def upload(
         staging_path: Path,
-        bucket: str = AWS_S3_BUCKET_NAME,
+        bucket: str = None,
         suite: str = "test",
         release: str = datetime.now().strftime("%Y-%m-%d"),
         file_name: str = None,
 ) -> str:
+    if bucket is None:
+        bucket = settings.AWS_S3_BUCKET_NAME
     if not file_name:
         file_name = uuid.uuid4().hex
     blob_s3_key = f"{suite}/{release}/{file_name}"

apps/api/src/services/celery_client.py

@@ -1,8 +1,8 @@
 from celery import Celery
-from src.config import CELERY_BROKER_URL, CELERY_RESULT_BACKEND
+from src.config import settings
 
 celery_app = Celery(
     "fastfetchbot_worker",
-    broker=CELERY_BROKER_URL,
-    backend=CELERY_RESULT_BACKEND,
+    broker=settings.CELERY_BROKER_URL,
+    backend=settings.CELERY_RESULT_BACKEND,
 )

apps/api/src/services/file_export/audio_transcribe/__init__.py

@@ -2,7 +2,7 @@
 
 from fastfetchbot_shared.services.file_export.audio_transcribe import AudioTranscribe as BaseAudioTranscribe
 from src.services.celery_client import celery_app
-from src.config import DOWNLOAD_VIDEO_TIMEOUT
+from src.config import settings
 
 
 class AudioTranscribe(BaseAudioTranscribe):
@@ -12,5 +12,5 @@ def __init__(self, audio_file: str):
         super().__init__(
             audio_file=audio_file,
             celery_app=celery_app,
-            timeout=DOWNLOAD_VIDEO_TIMEOUT,
+            timeout=settings.DOWNLOAD_VIDEO_TIMEOUT,
         )

apps/api/src/services/file_export/document_export/pdf_export.py

@@ -5,7 +5,7 @@
 import aiofiles.os
 
 from fastfetchbot_shared.services.file_export.pdf_export import PdfExport as BasePdfExport, wrap_html_string
-from src.config import DOWNLOAD_VIDEO_TIMEOUT, AWS_STORAGE_ON
+from src.config import settings
 from src.services.celery_client import celery_app
 from src.services.amazon.s3 import upload as upload_to_s3
 from fastfetchbot_shared.utils.logger import logger
@@ -27,13 +27,13 @@ def __init__(self, title: str, html_string: str = None):
             title=title,
             html_string=html_string,
             celery_app=celery_app,
-            timeout=DOWNLOAD_VIDEO_TIMEOUT,
+            timeout=settings.DOWNLOAD_VIDEO_TIMEOUT,
         )
 
     async def export(self) -> str:
         output_filename = await super().export()
 
-        if AWS_STORAGE_ON:
+        if settings.AWS_STORAGE_ON:
             local_filename = output_filename
             output_filename = await upload_file_to_s3(Path(output_filename))
             await aiofiles.os.remove(local_filename)

apps/api/src/services/file_export/video_download/__init__.py

@@ -4,7 +4,7 @@
 
 from fastfetchbot_shared.services.file_export.video_download import VideoDownloader as BaseVideoDownloader
 from src.services.celery_client import celery_app
-from src.config import DOWNLOAD_VIDEO_TIMEOUT
+from src.config import settings
 
 
 class VideoDownloader(BaseVideoDownloader):
@@ -25,7 +25,7 @@ def __init__(
             url=url,
             category=category,
             celery_app=celery_app,
-            timeout=DOWNLOAD_VIDEO_TIMEOUT,
+            timeout=settings.DOWNLOAD_VIDEO_TIMEOUT,
             data=data,
             download=download,
             audio_only=audio_only,

apps/api/src/services/inoreader/__init__.py

@@ -10,12 +10,7 @@
 from fastfetchbot_shared.utils.network import HEADERS
 from fastfetchbot_shared.utils.logger import logger
 from fastfetchbot_shared.utils.parse import get_html_text_length
-from src.config import (
-    INOREADER_APP_ID,
-    INOREADER_APP_KEY,
-    INOREADER_EMAIL,
-    INOREADER_PASSWORD,
-)
+from src.config import settings
 
 INOREADER_CONTENT_URL = "https://www.inoreader.com/reader/api/0/stream/contents/"
 TAG_PATH = "user/-/label/"
@@ -144,8 +139,8 @@ async def get_api_info(
             resp = await client.post(
                 INOREADER_LOGIN_URL,
                 params={
-                    "Email": INOREADER_EMAIL,
-                    "Passwd": INOREADER_PASSWORD,
+                    "Email": settings.INOREADER_EMAIL,
+                    "Passwd": settings.INOREADER_PASSWORD,
                 },
             )
             authorization = resp.text.split("\n")[2].split("=")[1]
@@ -156,8 +151,8 @@ async def get_api_info(
             params = params or {}
             params.update(
                 {
-                    "AppId": INOREADER_APP_ID,
-                    "AppKey": INOREADER_APP_KEY,
+                    "AppId": settings.INOREADER_APP_ID,
+                    "AppKey": settings.INOREADER_APP_KEY,
                 }
             )
             resp = await client.get(

apps/api/src/services/inoreader/process.py

@@ -2,7 +2,7 @@
 
 import httpx
 
-from src.config import TELEGRAM_BOT_CALLBACK_URL
+from src.config import settings
 from fastfetchbot_shared.models.url_metadata import UrlMetadata
 from src.services.inoreader import Inoreader
 from src.services.scrapers.common import InfoExtractService
@@ -19,7 +19,7 @@ async def _default_message_callback(metadata_item: dict, chat_id: Union[int, str
     """Default callback that sends via HTTP to the Telegram bot service."""
     async with httpx.AsyncClient() as client:
         await client.post(
-            f"{TELEGRAM_BOT_CALLBACK_URL}/send_message",
+            f"{settings.TELEGRAM_BOT_CALLBACK_URL}/send_message",
             json={"data": metadata_item, "chat_id": str(chat_id)},
             timeout=120,
         )