diff --git a/core/learninghouse/core/settings.py b/core/learninghouse/core/settings.py deleted file mode 100644 index d3503b5..0000000 --- a/core/learninghouse/core/settings.py +++ /dev/null @@ -1,150 +0,0 @@ -from functools import lru_cache -from pathlib import Path -from secrets import token_hex -from typing import Any, Dict, Optional, Union - -from pydantic import DirectoryPath -from pydantic_settings import BaseSettings, SettingsConfigDict - -from learninghouse import versions -from learninghouse.api.errors import ( - LearningHouseException, - LearningHouseValidationError, -) -from learninghouse.core.logging import LoggingLevelEnum -from learninghouse.models.base import EnumModel - -LICENSE_URL = "https://github.com/LearningHouseService/learninghouse/blob/main/LICENSE" - - -class ServiceSettings(BaseSettings): - debug: Optional[bool] = False - docs_url: str = "/docs" - openapi_file: str = "/learninghouse_api.json" - title: str = "learningHouse Service" - - host: str = "127.0.0.1" - port: int = 5000 - - workers: int = 5 - - reload: bool = False - base_url: str = '' - - environment: str = "production" - - config_directory: DirectoryPath = "./brains" - - logging_level: LoggingLevelEnum = LoggingLevelEnum.INFO - - jwt_secret: str = token_hex(16) - jwt_expire_minutes: int = 10 - - model_config = SettingsConfigDict( - env_file=".env", env_prefix="learninghouse_", ) - - @property - def fastapi_kwargs(self) -> Dict[str, Any]: - validation_error = LearningHouseValidationError - return { - "debug": self.debug, - "title": self.title, - "openapi_url": self.openapi_file, - "docs_url": None, - "redoc_url": None, - "version": versions.service, - "responses": { - validation_error.STATUS_CODE: validation_error.api_description(), - LearningHouseException.STATUS_CODE: LearningHouseException.api_description(), - }, - "license_info": {"name": "MIT License", "url": LICENSE_URL}, - } - - @property - def uvicorn_kwargs(self) -> Dict[str, Any]: - kwargs = { - "host": self.host, - "port": self.port, - "headers": [("server", f"LearningHouse Service {versions.service}")], - } - - if self.reload: - kwargs["reload"] = True - else: - kwargs["workers"] = self.workers - - return kwargs - - @property - def brains_directory(self) -> Path: - return Path(self.config_directory).absolute() - - @property - def base_url_calculated(self) -> str: - if self.base_url: - base_url = self.base_url - elif self.host in ("0.0.0.0", "127.0.0.1"): - base_url = "http://localhost" - else: - base_url = f"http://{self.host}" - - return f"{base_url}:{self.port}" - - @property - def documentation_url(self) -> Union[str, None]: - documentation_url = None - - if self.docs_url is not None: - documentation_url = self.base_url_calculated + self.docs_url - - return documentation_url - - @property - def openapi_url(self) -> str: - return self.base_url_calculated + self.openapi_file - - @property - def jwt_payload_claims(self) -> Dict[str, str]: - return {"audience": "LearningHouseAPI", "issuer": "LearningHouse Service"} - - -class DevelopmentSettings(ServiceSettings): - environment: str = "development" - debug: bool = True - reload: bool = True - title: str = "learningHouse Service - Development" - api_key_required: bool = False - - model_config = SettingsConfigDict( - env_file=".env", env_prefix="learninghouse_") - - -class ServiceEnvironment(EnumModel): - PROD = "production", ServiceSettings - DEV = "development", DevelopmentSettings - - def __init__(self, description: str, settings_class: ServiceSettings): - # pylint: disable=super-init-not-called - self._description: str = description - self._settings_class: ServiceSettings = settings_class - - @property - def description(self) -> str: - return self._description - - @property - def settings_class(self) -> ServiceSettings: - return self._settings_class - - -class ServiceBaseSettings(BaseSettings): - environment: Optional[ServiceEnvironment] = ServiceEnvironment.PROD - - model_config = SettingsConfigDict( - env_file=".env", env_prefix="learninghouse_", extra="ignore") - - -@lru_cache() -def service_settings() -> ServiceSettings: - environment = ServiceBaseSettings().environment - return environment.settings_class() diff --git a/core/learninghouse/core/settings/__init__.py b/core/learninghouse/core/settings/__init__.py new file mode 100644 index 0000000..66292d6 --- /dev/null +++ b/core/learninghouse/core/settings/__init__.py @@ -0,0 +1,7 @@ +from functools import lru_cache + +from learninghouse.core.settings.models import ServiceSettings + +@lru_cache() +def service_settings() -> ServiceSettings: + return ServiceSettings() diff --git a/core/learninghouse/core/settings/models.py b/core/learninghouse/core/settings/models.py new file mode 100644 index 0000000..aaa31f5 --- /dev/null +++ b/core/learninghouse/core/settings/models.py @@ -0,0 +1,174 @@ +from os import environ, listdir, path +from pathlib import Path +from secrets import token_hex +from typing import Any, Dict, Generator, Optional, Union + +from pydantic import BaseModel, DirectoryPath + +from learninghouse import versions +from learninghouse.api.errors import ( + LearningHouseException, + LearningHouseValidationError, +) +from learninghouse.core.logging import LoggingLevelEnum + +DOCKER_SECRETS_DIR = "/run/secrets" + +LICENSE_URL = "https://github.com/LearningHouseService/learninghouse/blob/main/LICENSE" + + +class ServiceSettings(BaseModel): + debug: Optional[bool] = False + docs_url: str = "/docs" + openapi_file: str = "/learninghouse_api.json" + title: str = "learningHouse Service" + + host: str = "127.0.0.1" + port: int = 5000 + + workers: int = 1 + + reload: bool = False + base_url: str = "" + + environment: str = "production" + + config_directory: DirectoryPath = "./brains" + + logging_level: LoggingLevelEnum = LoggingLevelEnum.INFO + + jwt_secret: str = token_hex(16) + jwt_expire_minutes: int = 10 + + def __init__(self, **data: dict[str, any]): + sources = [self._read_environment, self._read_dotenv, self._read_secrets] + data = self._parse_key_and_values(sources, data) + data = self.set_development_defaults(data) + + super().__init__(**data) + + def set_development_defaults(self, data: dict[str, any]) -> dict[str, any]: + if "environment" in data and data["environment"] == "development": + data = { + **{ + "debug": True, + "reload": True, + "title": "learningHouse Service - Development", + }, + **data, + } + + return data + + @property + def fastapi_kwargs(self) -> Dict[str, Any]: + validation_error = LearningHouseValidationError + return { + "debug": self.debug, + "title": self.title, + "openapi_url": self.openapi_file, + "docs_url": None, + "redoc_url": None, + "version": versions.service, + "responses": { + validation_error.STATUS_CODE: validation_error.api_description(), + LearningHouseException.STATUS_CODE: LearningHouseException.api_description(), + }, + "license_info": {"name": "MIT License", "url": LICENSE_URL}, + } + + @property + def uvicorn_kwargs(self) -> Dict[str, Any]: + kwargs = { + "host": self.host, + "port": self.port, + "headers": [("server", f"LearningHouse Service {versions.service}")], + } + + if self.reload: + kwargs["reload"] = True + else: + kwargs["workers"] = self.workers + + return kwargs + + @property + def brains_directory(self) -> Path: + return Path(self.config_directory).absolute() + + @property + def base_url_calculated(self) -> str: + if self.base_url: + base_url = self.base_url + elif self.host in ("0.0.0.0", "127.0.0.1"): + base_url = "http://localhost" + else: + base_url = f"http://{self.host}" + + return f"{base_url}:{self.port}" + + @property + def documentation_url(self) -> Union[str, None]: + documentation_url = None + + if self.docs_url is not None: + documentation_url = self.base_url_calculated + self.docs_url + + return documentation_url + + @property + def openapi_url(self) -> str: + return self.base_url_calculated + self.openapi_file + + @property + def jwt_payload_claims(self) -> Dict[str, str]: + return {"audience": "LearningHouseAPI", "issuer": "LearningHouse Service"} + + def _parse_key_and_values( + self, sources: list[callable], data: dict[str, any] + ) -> dict[str, any]: + for source in sources: + for key, value in source(): + key = key.lower().strip()[len("learninghouse_") :] # remove prefix + subkeys = key.split("__") # get nested structure + context = data + for subkey in subkeys[:-1]: + if subkey not in context: + context[subkey] = {} + context = context[subkey] + + context[subkeys[-1]] = ( + value.strip() + ) # Missing possibility to set nested json values + + return data + + @classmethod + def _read_environment(cls) -> Generator[tuple[str, str], any, any]: + for key, value in environ.items(): + if cls._has_prefix(key): + yield key, value + + @classmethod + def _read_secrets(cls) -> Generator[tuple[str, str], any, any]: + if path.exists(DOCKER_SECRETS_DIR) and path.isdir(DOCKER_SECRETS_DIR): + for filename in listdir(DOCKER_SECRETS_DIR): + if cls._has_prefix(filename): + with open( + path.join(DOCKER_SECRETS_DIR, filename), "r", encoding="utf-8" + ) as f: + yield filename, f.read() + + @classmethod + def _read_dotenv(cls) -> Generator[tuple[str, str], any, any]: + if path.exists(".env"): + with open(".env", "r", encoding="utf-8") as f: + for line in f.readlines(): + line = line.strip() + if cls._has_prefix(line) and "=" in line: + key, value = line.split("=", 1) + yield key, value + + @staticmethod + def _has_prefix(key: str) -> bool: + return key.lower().startswith("learninghouse_") diff --git a/core/requirements.txt b/core/requirements.txt index e585853..f4f5ebb 100644 --- a/core/requirements.txt +++ b/core/requirements.txt @@ -4,7 +4,6 @@ numpy==2.0.0 pandas==2.2.2 passlib==1.7.4 pydantic==2.8.2 -pydantic-settings==2.3.4 PyJWT==2.8.0 scikit-learn==1.5.0 uvicorn[standard]==0.30.1