MAESTRO: Create backend/config.py with Pydantic Settings and SQLite/in-process fallback

All 13 environment variables from the spec defined with proper defaults.
SQLite fallback when DATABASE_URL is unset, in-process queue flag when
REDIS_URL is unset, JWT_SECRET auto-generation, empty API_KEY normalization.
13 unit tests covering all configuration paths.
This commit is contained in:
John Lightner 2026-04-07 01:46:30 -05:00
parent 9e2961d648
commit 309bbacb5d
3 changed files with 183 additions and 1 deletions

View file

@ -17,7 +17,8 @@ Set up the PromptLooper repository, Docker infrastructure, and basic project ske
- [x] Create the multi-stage Dockerfile in docker/ that builds both backend and frontend into a single image. Stage 1: Node build for frontend (npm ci && npm run build). Stage 2: Python runtime with uvicorn, copying the built frontend assets. Include nginx.conf that serves the frontend and proxies /api and /ws to uvicorn. The image should work standalone with SQLite when no DATABASE_URL is provided. - [x] Create the multi-stage Dockerfile in docker/ that builds both backend and frontend into a single image. Stage 1: Node build for frontend (npm ci && npm run build). Stage 2: Python runtime with uvicorn, copying the built frontend assets. Include nginx.conf that serves the frontend and proxies /api and /ws to uvicorn. The image should work standalone with SQLite when no DATABASE_URL is provided.
> Created 3-stage Dockerfile: (1) frontend-build with Node 20 Alpine, (2) api stage with Python 3.12-slim + uvicorn + static assets for single-container mode, (3) web stage with nginx 1.27 Alpine for production compose. nginx.conf proxies /api/ and /health to the API, upgrades /ws/ connections for WebSocket. Also created: backend/requirements.txt, frontend scaffolding (package.json, vite.config.ts, tsconfig.json, index.html, App.tsx, Tailwind config), and placeholder alembic.ini/env.py for Dockerfile COPY. > Created 3-stage Dockerfile: (1) frontend-build with Node 20 Alpine, (2) api stage with Python 3.12-slim + uvicorn + static assets for single-container mode, (3) web stage with nginx 1.27 Alpine for production compose. nginx.conf proxies /api/ and /health to the API, upgrades /ws/ connections for WebSocket. Also created: backend/requirements.txt, frontend scaffolding (package.json, vite.config.ts, tsconfig.json, index.html, App.tsx, Tailwind config), and placeholder alembic.ini/env.py for Dockerfile COPY.
- [ ] Create backend/config.py using Pydantic Settings. Define all configuration from the Environment Variables table. Implement the SQLite fallback logic: when DATABASE_URL is not set, construct a SQLite URL pointing to DATA_DIR/promptlooper.db. When REDIS_URL is not set, set a flag for in-process mode. - [x] Create backend/config.py using Pydantic Settings. Define all configuration from the Environment Variables table. Implement the SQLite fallback logic: when DATABASE_URL is not set, construct a SQLite URL pointing to DATA_DIR/promptlooper.db. When REDIS_URL is not set, set a flag for in-process mode.
> Created backend/config.py with Pydantic Settings class defining all 13 env vars. SQLite fallback via `effective_database_url` property constructs sqlite:///DATA_DIR/promptlooper.db when DATABASE_URL is unset. `use_in_process_queue` property flags in-process mode when REDIS_URL is absent. JWT_SECRET auto-generates via `secrets.token_urlsafe(32)` when not provided. Empty API_KEY strings normalize to None. 13 tests in tests/test_config.py all passing.
- [ ] Create backend/models.py with all SQLAlchemy ORM models from the spec's Data Model section: User, Project, Experiment, Run, StageResult, Score, ResponseCache, and WebhookConfig. Include all fields, types, relationships, and indexes. Use UUID primary keys and JSONB for flexible fields. - [ ] Create backend/models.py with all SQLAlchemy ORM models from the spec's Data Model section: User, Project, Experiment, Run, StageResult, Score, ResponseCache, and WebhookConfig. Include all fields, types, relationships, and indexes. Use UUID primary keys and JSONB for flexible fields.

76
backend/config.py Normal file
View file

@ -0,0 +1,76 @@
"""PromptLooper configuration — Pydantic Settings loaded from environment."""
import secrets
from pathlib import Path
from pydantic import field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
extra="ignore",
)
# --- Database ---
database_url: str | None = None
# --- Redis ---
redis_url: str | None = None
# --- Server ---
host: str = "0.0.0.0"
port: int = 8400
# --- Auth ---
jwt_secret: str = ""
api_key: str | None = None
# --- Default LLM Endpoint ---
default_endpoint_url: str | None = None
default_endpoint_key: str | None = None
# --- Limits ---
max_concurrent_runs: int = 4
max_tokens_per_sweep: int = 0 # 0 = unlimited
# --- Storage ---
data_dir: str = "/data"
# --- MCP ---
mcp_enabled: bool = True
mcp_port: int = 8401
def model_post_init(self, __context: object) -> None:
# Auto-generate JWT secret if not provided
if not self.jwt_secret:
self.jwt_secret = secrets.token_urlsafe(32)
@property
def effective_database_url(self) -> str:
"""Return DATABASE_URL or construct a SQLite URL from DATA_DIR."""
if self.database_url:
return self.database_url
db_path = Path(self.data_dir) / "promptlooper.db"
return f"sqlite:///{db_path}"
@property
def is_sqlite(self) -> bool:
return self.effective_database_url.startswith("sqlite")
@property
def use_in_process_queue(self) -> bool:
"""When Redis is unavailable, use in-process task execution."""
return self.redis_url is None
@field_validator("api_key", mode="before")
@classmethod
def empty_string_to_none(cls, v: str | None) -> str | None:
if v is not None and v.strip() == "":
return None
return v
settings = Settings()

View file

@ -0,0 +1,105 @@
"""Tests for backend/config.py."""
import os
from unittest.mock import patch
import pytest
from pydantic_settings import BaseSettings
from config import Settings
class TestSettings:
"""Test the Settings configuration class."""
def _make_settings(self, **env_vars: str) -> Settings:
"""Create a Settings instance with specific env vars, ignoring .env file."""
with patch.dict(os.environ, env_vars, clear=False):
return Settings(_env_file=None)
def test_defaults(self) -> None:
s = self._make_settings()
assert s.database_url is None
assert s.redis_url is None
assert s.host == "0.0.0.0"
assert s.port == 8400
assert s.api_key is None
assert s.default_endpoint_url is None
assert s.default_endpoint_key is None
assert s.max_concurrent_runs == 4
assert s.max_tokens_per_sweep == 0
assert s.data_dir == "/data"
assert s.mcp_enabled is True
assert s.mcp_port == 8401
def test_jwt_secret_auto_generated(self) -> None:
s = self._make_settings()
assert len(s.jwt_secret) > 0
def test_jwt_secret_auto_generated_unique(self) -> None:
s1 = self._make_settings()
s2 = self._make_settings()
assert s1.jwt_secret != s2.jwt_secret
def test_jwt_secret_from_env(self) -> None:
s = self._make_settings(JWT_SECRET="my-secret-key")
assert s.jwt_secret == "my-secret-key"
def test_sqlite_fallback_when_no_database_url(self) -> None:
s = self._make_settings(DATA_DIR="/tmp/test")
url = s.effective_database_url
assert url.startswith("sqlite:///")
assert url.endswith("promptlooper.db")
assert "tmp" in url and "test" in url
assert s.is_sqlite is True
def test_postgres_when_database_url_set(self) -> None:
url = "postgresql://user:pass@localhost:5432/promptlooper"
s = self._make_settings(DATABASE_URL=url)
assert s.effective_database_url == url
assert s.is_sqlite is False
def test_in_process_queue_when_no_redis(self) -> None:
s = self._make_settings()
assert s.use_in_process_queue is True
def test_celery_queue_when_redis_set(self) -> None:
s = self._make_settings(REDIS_URL="redis://localhost:6379/0")
assert s.use_in_process_queue is False
assert s.redis_url == "redis://localhost:6379/0"
def test_empty_api_key_becomes_none(self) -> None:
s = self._make_settings(API_KEY="")
assert s.api_key is None
def test_whitespace_api_key_becomes_none(self) -> None:
s = self._make_settings(API_KEY=" ")
assert s.api_key is None
def test_valid_api_key_preserved(self) -> None:
s = self._make_settings(API_KEY="sk-test-123")
assert s.api_key == "sk-test-123"
def test_env_overrides(self) -> None:
s = self._make_settings(
HOST="127.0.0.1",
PORT="9000",
MAX_CONCURRENT_RUNS="8",
MAX_TOKENS_PER_SWEEP="100000",
MCP_ENABLED="false",
MCP_PORT="9001",
)
assert s.host == "127.0.0.1"
assert s.port == 9000
assert s.max_concurrent_runs == 8
assert s.max_tokens_per_sweep == 100000
assert s.mcp_enabled is False
assert s.mcp_port == 9001
def test_default_endpoint_config(self) -> None:
s = self._make_settings(
DEFAULT_ENDPOINT_URL="http://localhost:11434/v1",
DEFAULT_ENDPOINT_KEY="sk-key",
)
assert s.default_endpoint_url == "http://localhost:11434/v1"
assert s.default_endpoint_key == "sk-key"