Michael-Antony's picture
Fix: Parse DATABASE_URL to extract individual connection parameters for asyncpg
69c71e7
"""
Configuration settings for Tracker microservice.
Loads environment variables and provides application settings.
"""
import os
from typing import Optional, List
from pydantic import model_validator, Field
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
"""Application settings loaded from environment variables"""
# Application
APP_NAME: str = "Tracker Microservice"
APP_VERSION: str = "1.0.0"
DEBUG: bool = False
# MongoDB Configuration
MONGODB_URI: str = "mongodb://localhost:27017"
MONGODB_DB_NAME: str = "cuatrolabs"
# PostgreSQL Configuration
# Let Pydantic handle environment variables - don't use os.getenv() here!
# Use Field with alias to map DB_HOST -> POSTGRES_HOST, etc.
POSTGRES_HOST: str = Field(default="localhost", validation_alias="DB_HOST")
POSTGRES_PORT: int = Field(default=5432, validation_alias="DB_PORT")
POSTGRES_DB: str = Field(default="cuatrolabs", validation_alias="DB_NAME")
POSTGRES_USER: str = Field(default="postgres", validation_alias="DB_USER")
POSTGRES_PASSWORD: str = Field(default="", validation_alias="DB_PASSWORD")
POSTGRES_MIN_POOL_SIZE: int = 5
POSTGRES_MAX_POOL_SIZE: int = 20
POSTGRES_CONNECT_MAX_RETRIES: int = 20
POSTGRES_CONNECT_INITIAL_DELAY_MS: int = 500
POSTGRES_CONNECT_BACKOFF_MULTIPLIER: float = 1.5
POSTGRES_SSL_MODE: str = Field(default="disable", validation_alias="DB_SSLMODE")
POSTGRES_SSL_ROOT_CERT: Optional[str] = None
POSTGRES_SSL_CERT: Optional[str] = None
POSTGRES_SSL_KEY: Optional[str] = None
POSTGRES_URI: Optional[str] = None
@model_validator(mode='after')
def assemble_db_connection(self) -> 'Settings':
from urllib.parse import quote_plus, urlparse
# Prefer DATABASE_URL and DATABASE_URI
env_url = (os.getenv("DATABASE_URL") or os.getenv("DATABASE_URI") or "").strip()
if env_url:
self.POSTGRES_URI = env_url
print(f"[CONFIG] Using provided DATABASE_URL/URI")
# Parse the URL to extract individual components for asyncpg
try:
# Remove the +asyncpg suffix if present for parsing
parse_url = env_url.replace("postgresql+asyncpg://", "postgresql://")
parsed = urlparse(parse_url)
# Override individual settings from URL
if parsed.hostname:
self.POSTGRES_HOST = parsed.hostname
if parsed.port:
self.POSTGRES_PORT = parsed.port
if parsed.username:
self.POSTGRES_USER = parsed.username
if parsed.password:
self.POSTGRES_PASSWORD = parsed.password
if parsed.path and len(parsed.path) > 1:
self.POSTGRES_DB = parsed.path[1:] # Remove leading /
# Parse query parameters for SSL mode
if parsed.query:
from urllib.parse import parse_qs
params = parse_qs(parsed.query)
if 'sslmode' in params:
self.POSTGRES_SSL_MODE = params['sslmode'][0]
print(f"[CONFIG] Parsed DATABASE_URL:")
print(f"[CONFIG] Host: {self.POSTGRES_HOST}")
print(f"[CONFIG] Port: {self.POSTGRES_PORT}")
print(f"[CONFIG] Database: {self.POSTGRES_DB}")
print(f"[CONFIG] User: {self.POSTGRES_USER}")
print(f"[CONFIG] SSL Mode: {self.POSTGRES_SSL_MODE}")
except Exception as e:
print(f"[CONFIG] Warning: Failed to parse DATABASE_URL: {e}")
return self
# Build DSN from individual parts
if all([self.POSTGRES_USER, self.POSTGRES_PASSWORD, self.POSTGRES_HOST, self.POSTGRES_DB]):
protocol = os.getenv("DB_PROTOCOL", "postgresql+asyncpg")
# Ensure no spaces in connection components
user = self.POSTGRES_USER.strip()
host = self.POSTGRES_HOST.strip()
port = str(self.POSTGRES_PORT).strip()
db = self.POSTGRES_DB.strip()
self.POSTGRES_URI = f"{protocol}://{user}:{quote_plus(self.POSTGRES_PASSWORD)}@{host}:{port}/{db}"
print(f"[CONFIG] Built POSTGRES_URI from components")
print(f"[CONFIG] Protocol: {protocol}")
print(f"[CONFIG] User: {self.POSTGRES_USER}")
print(f"[CONFIG] Host: {self.POSTGRES_HOST}")
print(f"[CONFIG] Port: {self.POSTGRES_PORT}")
print(f"[CONFIG] Database: {self.POSTGRES_DB}")
print(f"[CONFIG] Password: {'SET' if self.POSTGRES_PASSWORD else 'EMPTY'}")
print(f"[CONFIG] SSL Mode: {self.POSTGRES_SSL_MODE}")
else:
self.POSTGRES_URI = None
print(f"[CONFIG] ERROR: Cannot build POSTGRES_URI - missing required components")
print(f"[CONFIG] POSTGRES_USER: {'SET' if self.POSTGRES_USER else 'MISSING'}")
print(f"[CONFIG] POSTGRES_PASSWORD: {'SET' if self.POSTGRES_PASSWORD else 'MISSING'}")
print(f"[CONFIG] POSTGRES_HOST: {'SET' if self.POSTGRES_HOST else 'MISSING'}")
print(f"[CONFIG] POSTGRES_DB: {'SET' if self.POSTGRES_DB else 'MISSING'}")
return self
# JWT Configuration
SECRET_KEY: str = "your-secret-key-change-in-production"
ALGORITHM: str = "HS256"
TOKEN_EXPIRATION_HOURS: int = 8
# Logging
LOG_LEVEL: str = "INFO"
# CORS
CORS_ORIGINS: List[str] = [
"http://localhost:3000",
"http://localhost:8000",
"http://localhost:8003",
]
# Pydantic v2 config
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=True,
extra="allow", # allows extra environment variables without error
# Priority order (highest to lowest):
# 1. OS environment variables (Docker, shell exports)
# 2. .env file (local development)
# 3. Default values (fallback)
env_prefix="", # No prefix, use exact names
)
# Global settings instance
settings = Settings()