""" Application settings. """
from typing import Optional, Any, Dict, ClassVar, List
import enum
from pathlib import Path
import logging
import json
from pydantic import BaseSettings
[docs]
LOGGER = logging.getLogger(__name__)
[docs]
class LogLevel(str, enum.Enum): # noqa: WPS600
"""Possible log levels."""
[docs]
class RMSettings(BaseSettings): # pylint: disable=too-few-public-methods
"""
Application settings.
These parameters can be configured
with environment variables.
"""
[docs]
class Config: # pylint: disable=too-few-public-methods
"""Configuration of settings."""
[docs]
env_file_encoding = "utf-8"
[docs]
host: str = "127.0.0.1"
# quantity of workers for uvicorn
# Enable uvicorn reloading
# Current environment
[docs]
environment: str = "dev"
# Set log_level (str) and log_level_int (int) for later use
# if log_level is not set, then log level will be DEBUG
[docs]
log_level: LogLevel = LogLevel.DEBUG
[docs]
log_level_int: int = logging.DEBUG
if log_level == "INFO":
log_level_int = logging.INFO
elif log_level == "WARNING":
log_level_int = logging.WARNING
elif log_level == "ERROR":
log_level_int = logging.ERROR
elif log_level == "FATAL":
log_level_int = logging.FATAL
# Manifest file from kraftwerk
[docs]
integration_api_port: int = 4625
[docs]
kraftwerk_manifest_path: str = "/pvarki/kraftwerk-rasenmaeher-init.json"
[docs]
kraftwerk_manifest_bool: bool = False
[docs]
kraftwerk_manifest_dict: Dict[Any, Any] = {}
[docs]
integration_api_timeout: float = 3.0
# Api access management
[docs]
api_healthcheck_proto: str = "http://"
[docs]
api_healthcheck_url: str = "/api/v1/healthcheck"
# Sentry's configuration.
[docs]
sentry_dsn: Optional[str] = None
[docs]
sentry_sample_rate: float = 1.0
# Cfssl configuration
[docs]
cfssl_host: str = "http://127.0.0.1"
[docs]
cfssl_port: str = "8888"
[docs]
ocsprest_host: str = "http://127.0.0.1"
[docs]
ocsprest_port: str = "8887"
[docs]
cfssl_timeout: float = 2.5
[docs]
persistent_data_dir = "/data/persistent"
# mtls
[docs]
mtls_client_cert_path: Optional[str] = None
[docs]
mtls_client_key_path: Optional[str] = None
[docs]
mtls_client_cert_cn = "rasenmaeher"
# Keycloak configuration.
[docs]
keycloak_server_url: Optional[str] = None
[docs]
keycloak_client_id: Optional[str] = None
[docs]
keycloak_realm_name: Optional[str] = None
[docs]
keycloak_client_secret: Optional[str] = None
# LDAP configuration
[docs]
ldap_conn_string: Optional[str] = None
[docs]
ldap_username: Optional[str] = None
[docs]
ldap_client_secret: Optional[str] = None
# Tilauspalvelu integration
[docs]
tilauspalvelu_jwt: str = "https://tilaa.pvarki.fi/api/v1/config/jwtPublicKey.pem"
[docs]
kraftwerk_announce: Optional[str] = None # When KRAFTWERK actually exists
[docs]
kraftwerk_timeout: float = 2.0
[docs]
_singleton: ClassVar[Optional["RMSettings"]] = None
@classmethod
[docs]
def singleton(cls) -> "RMSettings":
"""Return singleton"""
if not RMSettings._singleton:
RMSettings._singleton = RMSettings()
return RMSettings._singleton
def __init__(self, *args: List[Any], **kwargs: Dict[str, Any]) -> None:
super().__init__(*args, **kwargs) # type: ignore[arg-type]
# FIXME: When the switchme_to_singleton_call has been removed call load_manifest here
[docs]
def load_manifest(self) -> None:
"""Load the kraftwerk manifest file"""
if self.kraftwerk_manifest_bool:
return
pth = Path(self.kraftwerk_manifest_path)
if not pth.exists():
raise ValueError(f"{self.kraftwerk_manifest_path} does not exist")
self.kraftwerk_manifest_dict = json.loads(pth.read_text(encoding="utf-8"))
self.kraftwerk_manifest_bool = True
@property
[docs]
def valid_product_cns(self) -> List[str]:
"""Get valid CNs for productapi certs"""
self.load_manifest()
return [product["certcn"] for product in self.kraftwerk_manifest_dict["products"].values()]
[docs]
switchme_to_singleton_call = RMSettings.singleton() # pylint: disable=C0103