[rewrite] Major changes to code - switched to postgres, added initial backend docker files
This commit is contained in:
Thastertyn committed
1 parent
c7e20fc935
commit
71e916586e
52 files changed
+783
-537
No files matched your search
@@ -1,3 +0,0 @@
|
||||
from .config import *
|
||||
|
||||
__all__ = [*config.__all__]
|
||||
+85
-183
@@ -1,205 +1,107 @@
|
||||
import os
|
||||
import secrets
|
||||
import warnings
|
||||
import logging
|
||||
import re
|
||||
from typing import Any, Type, Optional, Callable
|
||||
from typing import Annotated, Any, Literal
|
||||
|
||||
import dotenv
|
||||
|
||||
from .exceptions import ConfigError
|
||||
from pydantic import (
|
||||
AnyUrl,
|
||||
BeforeValidator,
|
||||
EmailStr,
|
||||
PostgresDsn,
|
||||
computed_field,
|
||||
model_validator,
|
||||
)
|
||||
from pydantic_core import MultiHostUrl
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
from typing_extensions import Self
|
||||
|
||||
|
||||
class EnvConfigField():
|
||||
"""
|
||||
Represents a single config field to be parsed from the environment.
|
||||
def parse_cors(v: Any) -> list[str] | str:
|
||||
if isinstance(v, str) and not v.startswith("["):
|
||||
return [i.strip() for i in v.split(",")]
|
||||
elif isinstance(v, list | str):
|
||||
return v
|
||||
raise ValueError(v)
|
||||
|
||||
:param env_name: Name of the environment variable
|
||||
:param data_type: Expected data type (int, str, etc.)
|
||||
:param default: Default value if not set in environment
|
||||
:param required: Whether the variable is required
|
||||
:param choices: List of allowed values (for enums)
|
||||
:param min_value: Minimum value (for integers)
|
||||
:param max_value: Maximum value (for integers)
|
||||
:param regex: Regex pattern (for string validation)
|
||||
:param parser: Custom parser function for complex validation
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
env_name: str,
|
||||
data_type: Type[Any] = str,
|
||||
default: Any = None,
|
||||
required: bool = False,
|
||||
choices: Optional[list[Any]] = None,
|
||||
min_value: Optional[int] = None,
|
||||
max_value: Optional[int] = None,
|
||||
regex: Optional[str] = None,
|
||||
parser: Optional[Callable[[str], Any]] = None,
|
||||
):
|
||||
self.env_value = None
|
||||
self.env_name = env_name
|
||||
self.data_type = data_type
|
||||
self.default = default
|
||||
self.required = required
|
||||
self.choices = choices
|
||||
self.min_value = min_value
|
||||
self.max_value = max_value
|
||||
self.regex = re.compile(regex) if regex else None
|
||||
self.parser = parser
|
||||
class Settings(BaseSettings):
|
||||
model_config = SettingsConfigDict(
|
||||
env_file="../.env",
|
||||
env_ignore_empty=True,
|
||||
extra="ignore",
|
||||
)
|
||||
PORT: int = 8000
|
||||
SECRET_KEY: str = secrets.token_urlsafe(32)
|
||||
|
||||
def load(self):
|
||||
"""Loads the value from the environment and validates it."""
|
||||
raw_value = os.environ.get(self.env_name)
|
||||
VERBOSITY: Literal[*logging.getLevelNamesMapping().keys()] = "DEBUG"
|
||||
|
||||
if self.required and (raw_value is None or raw_value == ""):
|
||||
raise ConfigError(
|
||||
"Field %(env_name)s is required, but was not provided in the environment / .env",
|
||||
env_name=self.env_name,
|
||||
)
|
||||
# 60 minutes * 24 hours * 8 days = 8 days
|
||||
ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8
|
||||
FRONTEND_HOST: str
|
||||
ENVIRONMENT: Literal["local", "staging", "production"] = "local"
|
||||
|
||||
self.env_value = self.validate(raw_value)
|
||||
|
||||
def validate(self, value: Any) -> Any:
|
||||
"""Validates the value based on provided constraints."""
|
||||
|
||||
# Use the default value if available and skip validation
|
||||
if (value is None or value == "") and self.default is not None:
|
||||
return self.default
|
||||
|
||||
# Use custom parser if provided
|
||||
if self.parser:
|
||||
value = self.parser(value)
|
||||
|
||||
# Convert value to expected data type
|
||||
if self.data_type == str:
|
||||
value = self.__parse_to_str(value)
|
||||
elif self.data_type == int:
|
||||
value = self.__parse_to_int(value)
|
||||
elif self.data_type == bool:
|
||||
value = self.__parse_to_bool(value)
|
||||
|
||||
# Validate allowed choices
|
||||
if self.choices and value not in self.choices:
|
||||
raise ConfigError(
|
||||
"Value for %(env_name)s - %(value)s is not one of the following allowed values: %(allowed_values)s",
|
||||
env_name=self.env_name,
|
||||
value=value,
|
||||
allowed_values=", ".join(map(str, self.choices)),
|
||||
)
|
||||
|
||||
return value
|
||||
BACKEND_CORS_ORIGINS: Annotated[
|
||||
list[AnyUrl] | str, BeforeValidator(parse_cors)
|
||||
] = []
|
||||
|
||||
@computed_field
|
||||
@property
|
||||
def value(self):
|
||||
return self.env_value
|
||||
def all_cors_origins(self) -> list[str]:
|
||||
return [str(origin).rstrip("/") for origin in self.BACKEND_CORS_ORIGINS] + [self.FRONTEND_HOST]
|
||||
|
||||
@value.setter
|
||||
def value(self, new):
|
||||
raise TypeError("Cannot modify")
|
||||
POSTGRES_SERVER: str
|
||||
POSTGRES_PORT: int = 5432
|
||||
POSTGRES_USER: str
|
||||
POSTGRES_PASSWORD: str = ""
|
||||
POSTGRES_DB: str = ""
|
||||
|
||||
def __parse_to_str(self, value) -> str:
|
||||
"""Parses and validates string values."""
|
||||
value = str(value)
|
||||
@computed_field
|
||||
@property
|
||||
def SQLALCHEMY_DATABASE_URI(self) -> PostgresDsn:
|
||||
return MultiHostUrl.build(
|
||||
scheme='postgresql+psycopg2',
|
||||
username=self.POSTGRES_USER,
|
||||
password=self.POSTGRES_PASSWORD,
|
||||
host=self.POSTGRES_SERVER,
|
||||
port=self.POSTGRES_PORT,
|
||||
path=self.POSTGRES_DB,
|
||||
)
|
||||
|
||||
if self.regex and not self.regex.match(value):
|
||||
raise ConfigError(
|
||||
"Value for %(env_name)s - %(value)s does not match the expected pattern: %(regex)s",
|
||||
env_name=self.env_name,
|
||||
value=value,
|
||||
regex=self.regex.pattern,
|
||||
)
|
||||
SMTP_TLS: bool = True
|
||||
SMTP_SSL: bool = False
|
||||
SMTP_PORT: int = 587
|
||||
SMTP_HOST: str | None = None
|
||||
SMTP_USER: str | None = None
|
||||
SMTP_PASSWORD: str | None = None
|
||||
EMAILS_FROM_EMAIL: EmailStr | None = None
|
||||
EMAILS_FROM_NAME: EmailStr | None = None
|
||||
|
||||
return value
|
||||
EMAIL_RESET_TOKEN_EXPIRE_HOURS: int = 48
|
||||
|
||||
def __parse_to_int(self, value) -> int:
|
||||
"""Parses and validates integer values."""
|
||||
if not str(value).isdigit():
|
||||
raise ConfigError(
|
||||
"Value for %(env_name)s - %(value)s must be a numeric integer",
|
||||
env_name=self.env_name,
|
||||
value=value,
|
||||
)
|
||||
@computed_field
|
||||
@property
|
||||
def emails_enabled(self) -> bool:
|
||||
return bool(self.SMTP_HOST and self.EMAILS_FROM_EMAIL)
|
||||
|
||||
value = int(value)
|
||||
EMAIL_TEST_USER: EmailStr = "test@example.com"
|
||||
FIRST_SUPERUSER: EmailStr
|
||||
FIRST_SUPERUSER_PASSWORD: str
|
||||
|
||||
if self.min_value is not None and value < self.min_value:
|
||||
raise ConfigError(
|
||||
"Value for %(env_name)s - %(value)d is too small. Must be greater than or equal to %(min_value)d",
|
||||
env_name=self.env_name,
|
||||
value=value,
|
||||
min_value=self.min_value,
|
||||
)
|
||||
def _check_default_secret(self, var_name: str, value: str | None) -> None:
|
||||
if value == "changethis":
|
||||
message = f'The value of {var_name} is "changethis", '"for security, please change it, at least for deployments."
|
||||
if self.ENVIRONMENT == "local":
|
||||
warnings.warn(message, stacklevel=1)
|
||||
else:
|
||||
raise ValueError(message)
|
||||
|
||||
if self.max_value is not None and value > self.max_value:
|
||||
raise ConfigError(
|
||||
"Value for %(env_name)s - %(value)d is too large. Must be smaller than or equal to %(max_value)d",
|
||||
env_name=self.env_name,
|
||||
value=value,
|
||||
max_value=self.max_value,
|
||||
)
|
||||
@model_validator(mode="after")
|
||||
def _enforce_non_default_secrets(self) -> Self:
|
||||
self._check_default_secret("SECRET_KEY", self.SECRET_KEY)
|
||||
self._check_default_secret("POSTGRES_PASSWORD", self.POSTGRES_PASSWORD)
|
||||
self._check_default_secret("FIRST_SUPERUSER_PASSWORD", self.FIRST_SUPERUSER_PASSWORD)
|
||||
|
||||
return value
|
||||
|
||||
def __parse_to_bool(self, value):
|
||||
value = str(value).lower()
|
||||
|
||||
true_values = ["1", "true", "yes"]
|
||||
false_values = ["0", "false", "no"]
|
||||
|
||||
if value in true_values:
|
||||
return True
|
||||
if value in false_values:
|
||||
return False
|
||||
|
||||
raise ConfigError("Value for %(env_name)s - %(value)s cannot be converted to a true / false value",
|
||||
env_name=self.env_name,
|
||||
value=value)
|
||||
return self
|
||||
|
||||
|
||||
class EnvConfig():
|
||||
|
||||
_instance = None
|
||||
_initialized = False
|
||||
|
||||
def __new__(cls):
|
||||
print("Called __new__")
|
||||
if cls._instance is None:
|
||||
cls._instance = super().__new__(cls)
|
||||
return cls._instance
|
||||
|
||||
def __init__(self):
|
||||
if EnvConfig._initialized:
|
||||
return
|
||||
|
||||
self.logger = logging.getLogger(__name__)
|
||||
|
||||
self.fields = {}
|
||||
|
||||
EnvConfig._initialized = True
|
||||
|
||||
def add_field(self, name: str, field: EnvConfigField):
|
||||
if name in self.fields:
|
||||
return
|
||||
|
||||
self.fields[name] = field
|
||||
|
||||
def load_config(self):
|
||||
dotenv.load_dotenv()
|
||||
|
||||
for field in self.fields.values():
|
||||
try:
|
||||
field.load()
|
||||
except ConfigError as e:
|
||||
self.logger.error("Configuration error: %s", e)
|
||||
raise
|
||||
|
||||
def __getitem__(self, key: str):
|
||||
return self.fields[key].value
|
||||
|
||||
def __contains__(self, key: str) -> bool:
|
||||
return key in self.fields
|
||||
|
||||
def to_dict(self):
|
||||
return {attr: getattr(self, attr) for attr in self.fields}
|
||||
|
||||
|
||||
__all__ = ["EnvConfig"]
|
||||
settings = Settings()
|
||||
@@ -0,0 +1,13 @@
|
||||
class SwagShopError(Exception):
|
||||
"""
|
||||
Generic error to be raised throughout the entire app.
|
||||
All other custom errors extend this class
|
||||
"""
|
||||
|
||||
|
||||
class RepositoryError(SwagShopError):
|
||||
"""Generic error occuring in a repository"""
|
||||
|
||||
|
||||
class ServiceError(SwagShopError):
|
||||
"""Generic error occuring in a service"""
|
||||
@@ -1,12 +0,0 @@
|
||||
class SwagShopError(Exception):
|
||||
def __init__(self, message):
|
||||
super().__init__(message)
|
||||
self.message = message
|
||||
|
||||
class ConfigError(SwagShopError):
|
||||
def __init__(self, message, **kwargs):
|
||||
formatted_message = message % kwargs
|
||||
super().__init__(formatted_message)
|
||||
self.message = formatted_message
|
||||
|
||||
__all__ = ["SwagShopError", "ConfigError"]
|
||||
@@ -5,7 +5,7 @@ import jwt
|
||||
|
||||
from passlib.context import CryptContext
|
||||
|
||||
from app.core.config import EnvConfig
|
||||
from app.core.config import settings
|
||||
|
||||
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
||||
|
||||
@@ -14,11 +14,9 @@ ALGORITHM = "HS256"
|
||||
|
||||
|
||||
def create_access_token(subject: str | Any, expires_delta: timedelta) -> str:
|
||||
config = EnvConfig()
|
||||
|
||||
expire = datetime.now(timezone.utc) + expires_delta
|
||||
to_encode = {"exp": expire, "sub": str(subject)}
|
||||
encoded_jwt = jwt.encode(to_encode, config["secret_key"], algorithm=ALGORITHM)
|
||||
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM)
|
||||
return encoded_jwt
|
||||
|
||||
|
||||
|
||||
Reference in new issue
Block a user