fix(config): validation logic

This commit is contained in:
sankalp committed 2025-04-22 05:05:20 +05:45
1 parent 369714083b
commit ac54ecd66c
1 file changed
+69 -18
+69 -18
View File
@@ -16,25 +16,29 @@ from pydantic import BaseModel, Field, ValidationError, field_validator, model_v
from dotenv import load_dotenv from dotenv import load_dotenv
import logging import logging
from functools import lru_cache from functools import lru_cache
from django.core.exceptions import ValidationError
# Configure logging # Configure logging
logging.basicConfig(level=logging.INFO) logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Configuration version # Configuration version for tracking changes
CONFIG_VERSION = "1.0.0" CONFIG_VERSION = "1.0.0"
def parse_optional_int(value: Any) -> Optional[int]: def parse_optional_int(value: Any) -> Optional[int]:
"""Parse a value into an optional integer.""" """Parse a value into an optional integer."""
if value is None or value == '': if value is None or value == "":
return None return None
try: try:
return int(value) return int(value)
except (ValueError, TypeError): except (ValueError, TypeError):
return None return None
class DatabaseConfig(BaseModel): class DatabaseConfig(BaseModel):
"""Database configuration settings.""" """Database configuration settings."""
engine: str = Field(default="django.db.backends.sqlite3") engine: str = Field(default="django.db.backends.sqlite3")
name: str = Field(default="db.sqlite3") name: str = Field(default="db.sqlite3")
timeout: int = Field(default=20) timeout: int = Field(default=20)
@@ -43,23 +47,27 @@ class DatabaseConfig(BaseModel):
user: Optional[str] = None user: Optional[str] = None
password: Optional[str] = None password: Optional[str] = None
@model_validator(mode='before') @model_validator(mode="before")
@classmethod @classmethod
def validate_optional_fields(cls, values: Dict[str, Any]) -> Dict[str, Any]: def validate_optional_fields(cls, values: Dict[str, Any]) -> Dict[str, Any]:
if 'port' in values: if "port" in values:
values['port'] = parse_optional_int(values['port']) values["port"] = parse_optional_int(values["port"])
return values return values
class CacheConfig(BaseModel): class CacheConfig(BaseModel):
"""Cache configuration settings.""" """Cache configuration settings."""
use_redis: bool = Field(default=False) use_redis: bool = Field(default=False)
redis_url: str = Field(default="redis://localhost:6379/1") redis_url: str = Field(default="redis://localhost:6379/1")
timeout: int = Field(default=300) timeout: int = Field(default=300)
key_prefix: str = Field(default="sms") key_prefix: str = Field(default="sms")
max_connections: int = Field(default=1000) max_connections: int = Field(default=1000)
class EmailConfig(BaseModel): class EmailConfig(BaseModel):
"""Email configuration settings.""" """Email configuration settings."""
backend: str = Field(default="django.core.mail.backends.console.EmailBackend") backend: str = Field(default="django.core.mail.backends.console.EmailBackend")
default_from_email: str = Field(default="noreply@studentmanagementsystem.com") default_from_email: str = Field(default="noreply@studentmanagementsystem.com")
host: Optional[str] = None host: Optional[str] = None
@@ -68,15 +76,17 @@ class EmailConfig(BaseModel):
password: Optional[str] = None password: Optional[str] = None
use_tls: bool = Field(default=True) use_tls: bool = Field(default=True)
@model_validator(mode='before') @model_validator(mode="before")
@classmethod @classmethod
def validate_optional_fields(cls, values: Dict[str, Any]) -> Dict[str, Any]: def validate_optional_fields(cls, values: Dict[str, Any]) -> Dict[str, Any]:
if 'port' in values: if "port" in values:
values['port'] = parse_optional_int(values['port']) values["port"] = parse_optional_int(values["port"])
return values return values
class FirebaseConfig(BaseModel): class FirebaseConfig(BaseModel):
"""Firebase configuration settings.""" """Firebase configuration settings."""
api_key: str api_key: str
auth_domain: str auth_domain: str
project_id: str project_id: str
@@ -85,20 +95,26 @@ class FirebaseConfig(BaseModel):
app_id: str app_id: str
measurement_id: str measurement_id: str
class SMSConfig(BaseModel): class SMSConfig(BaseModel):
"""SMS configuration settings.""" """SMS configuration settings."""
api_key: str api_key: str
sender_id: str = Field(default="SMSSYS") sender_id: str = Field(default="SMSSYS")
class GunicornConfig(BaseModel): class GunicornConfig(BaseModel):
"""Gunicorn configuration settings.""" """Gunicorn configuration settings."""
workers: int = Field(default=4) workers: int = Field(default=4)
threads: int = Field(default=2) threads: int = Field(default=2)
timeout: int = Field(default=30) timeout: int = Field(default=30)
keepalive: int = Field(default=2) keepalive: int = Field(default=2)
class Config(BaseModel): class Config(BaseModel):
"""Main configuration model.""" """Main configuration model."""
version: str = Field(default=CONFIG_VERSION) version: str = Field(default=CONFIG_VERSION)
debug: bool = Field(default=False) debug: bool = Field(default=False)
secret_key: str secret_key: str
@@ -114,6 +130,7 @@ class Config(BaseModel):
env_file = ".env" env_file = ".env"
env_file_encoding = "utf-8" env_file_encoding = "utf-8"
@lru_cache() @lru_cache()
def get_config() -> Config: def get_config() -> Config:
""" """
@@ -150,8 +167,12 @@ def get_config() -> Config:
max_connections=int(os.getenv("CACHE_MAX_CONNECTIONS", "1000")), max_connections=int(os.getenv("CACHE_MAX_CONNECTIONS", "1000")),
), ),
email=EmailConfig( email=EmailConfig(
backend=os.getenv("EMAIL_BACKEND", "django.core.mail.backends.console.EmailBackend"), backend=os.getenv(
default_from_email=os.getenv("DEFAULT_FROM_EMAIL", "noreply@studentmanagementsystem.com"), "EMAIL_BACKEND", "django.core.mail.backends.console.EmailBackend"
),
default_from_email=os.getenv(
"DEFAULT_FROM_EMAIL", "noreply@studentmanagementsystem.com"
),
host=os.getenv("EMAIL_HOST"), host=os.getenv("EMAIL_HOST"),
port=os.getenv("EMAIL_PORT"), port=os.getenv("EMAIL_PORT"),
username=os.getenv("EMAIL_USERNAME"), username=os.getenv("EMAIL_USERNAME"),
@@ -181,11 +202,37 @@ def get_config() -> Config:
# Validate required fields # Validate required fields
if not config.secret_key: if not config.secret_key:
raise ValidationError("DJANGO_SECRET_KEY is required") # Check if running tests and provide fallback for testing
import sys
if "test" in sys.argv:
config.secret_key = "django-insecure-test-key-only-for-testing"
else:
from django.core.exceptions import ValidationError
raise ValidationError(f"DJANGO_SECRET_KEY is required")
if not config.firebase.api_key: if not config.firebase.api_key:
raise ValidationError("FIREBASE_API_KEY is required") # Check if running tests
import sys
if "test" in sys.argv:
config.firebase.api_key = "test-firebase-api-key"
else:
from django.core.exceptions import ValidationError
raise ValidationError(f"FIREBASE_API_KEY is required")
if not config.sms.api_key: if not config.sms.api_key:
raise ValidationError("SMS_API_KEY is required") # Check if running tests
import sys
if "test" in sys.argv:
config.sms.api_key = "test-sms-api-key"
else:
from django.core.exceptions import ValidationError
raise ValidationError(f"SMS_API_KEY is required")
return config return config
@@ -196,6 +243,7 @@ def get_config() -> Config:
logger.error(f"Error loading configuration: {e}") logger.error(f"Error loading configuration: {e}")
raise raise
def get_django_settings() -> Dict[str, Any]: def get_django_settings() -> Dict[str, Any]:
""" """
Convert the configuration to Django settings format. Convert the configuration to Django settings format.
@@ -219,17 +267,20 @@ def get_django_settings() -> Dict[str, Any]:
"PASSWORD": config.database.password, "PASSWORD": config.database.password,
"OPTIONS": { "OPTIONS": {
"timeout": config.database.timeout, "timeout": config.database.timeout,
} },
} }
}, },
"CACHES": { "CACHES": {
"default": { "default": {
"BACKEND": "django.core.cache.backends.redis.RedisCache" if config.cache.use_redis "BACKEND": "django.core.cache.backends.redis.RedisCache"
else "django.core.cache.backends.locmem.LocMemCache", if config.cache.use_redis
"LOCATION": config.cache.redis_url if config.cache.use_redis else "unique-snowflake", else "django.core.cache.backends.locmem.LocMemCache",
"LOCATION": config.cache.redis_url
if config.cache.use_redis
else "unique-snowflake",
"OPTIONS": { "OPTIONS": {
"retry_on_timeout": True, "retry_on_timeout": True,
"max_connections": config.cache.max_connections "max_connections": config.cache.max_connections,
}, },
"KEY_PREFIX": config.cache.key_prefix, "KEY_PREFIX": config.cache.key_prefix,
"TIMEOUT": config.cache.timeout, "TIMEOUT": config.cache.timeout,