from copy import deepcopy
from enum import Enum, IntEnum
from pathlib import Path
from typing import Annotated, Optional, TypedDict, Union, cast
import yaml
from glom import glom
from pydantic import BaseModel, ConfigDict, Field, PositiveInt, ValidationError
from controller import (
COMPOSE_FILE_VERSION,
CONFS_DIR,
PLACEHOLDER,
EnvType,
log,
print_and_exit,
)
PROJECTS_DEFAULTS_FILE = Path("projects_defaults.yaml")
PROJECTS_PROD_DEFAULTS_FILE = Path("projects_prod_defaults.yaml")
PROJECT_CONF_FILENAME = Path("project_configuration.yaml")
[docs]
class Project(TypedDict):
title: str
description: str
keywords: str
rapydo: str
version: str
extends: Optional[str]
extends_from: Optional[str]
[docs]
class Submodule(TypedDict):
online_url: str
branch: Optional[str]
_if: str
[docs]
class Variables(TypedDict):
submodules: dict[str, Submodule]
roles: dict[str, str]
env: dict[str, EnvType]
# total is False because of .projectrc and project_configuration
# But should be total=True in case of projects_defaults
[docs]
class Configuration(TypedDict, total=False):
project: Project
tags: dict[str, str]
variables: Variables
[docs]
class FRONTEND_FRAMEWORK_VALUES(Enum):
nofrontend = "nofrontend"
angular = "angular"
[docs]
class FRONTEND_BUILD_MODE_VALUES(Enum):
angular = "angular"
angular_test = "angular-test"
[docs]
class DOCKER_LOGGING_DRIVERS(Enum):
json_file = "json-file"
syslog = "syslog"
local = "local"
[docs]
class LOG_LEVEL_VALUES(Enum):
DEBUG = "DEBUG"
INFO = "INFO"
WARNING = "WARNING"
ERROR = "ERROR"
CRITICAL = "CRITICAL"
[docs]
class APP_MODE_VALUES(Enum):
development = "development"
production = "production"
test = "test"
[docs]
class BACKEND_PYTHON_VERSION_VALUES(Enum):
py312 = "v3.12"
py311 = "v3.11"
py310 = "v3.10"
py39 = "v3.9"
[docs]
class PYTHONMALLOC_VALUES(Enum):
empty = ""
debug = "debug"
malloc = "malloc"
pymalloc = "pymalloc"
[docs]
class AUTH_SERVICE_VALUES(Enum):
no = "NO_AUTHENTICATION"
postgres = "sqlalchemy"
neo4j = "neo4j"
[docs]
class ALCHEMY_DBTYPE_VALUES(Enum):
postgresql = "postgresql"
mysql = "mysql+pymysql"
[docs]
class CELERY_BROKER_VALUES(Enum):
RABBIT = "RABBIT"
REDIS = "REDIS"
[docs]
class CELERY_BACKEND_VALUES(Enum):
RABBIT = "RABBIT"
REDIS = "REDIS"
[docs]
class CELERY_POOL_MODE_VALUES(Enum):
prefork = "prefork"
eventlet = "eventlet"
gevent = "gevent"
thread = "thread"
solo = "solo"
[docs]
class FLOWER_PROTOCOL_VALUES(Enum):
http = "http"
https = "https"
[docs]
class NEO4J_BOLT_TLS_LEVEL_VALUES(Enum):
REQUIRED = "REQUIRED"
OPTIONAL = "OPTIONAL"
DISABLED = "DISABLED"
[docs]
class SPINNER_TYPES(Enum):
BALL_8BITS = "ball-8bits"
BALL_ATOM = "ball-atom"
BALL_BEAT = "ball-beat"
BALL_CIRCUS = "ball-circus"
BALL_CLIMBING_DOT = "ball-climbing-dot"
BALL_CLIP_ROTATE = "ball-clip-rotate"
BALL_CLIP_ROTATE_MULTIPLE = "ball-clip-rotate-multiple"
BALL_CLIP_ROTATE_PULSE = "ball-clip-rotate-pulse"
BALL_ELASTIC_DOTS = "ball-elastic-dots"
BALL_FALL = "ball-fall"
BALL_FUSSION = "ball-fussion"
BALL_GRID_BEAT = "ball-grid-beat"
BALL_GRID_PULSE = "ball-grid-pulse"
BALL_NEWTON_CRADLE = "ball-newton-cradle"
BALL_PULSE = "ball-pulse"
BALL_PULSE_RISE = "ball-pulse-rise"
BALL_PULSE_SYNC = "ball-pulse-sync"
BALL_ROTATE = "ball-rotate"
BALL_RUNNING_DOTS = "ball-running-dots"
BALL_SCALE = "ball-scale"
BALL_SCALE_MULTIPLE = "ball-scale-multiple"
BALL_SCALE_PULSE = "ball-scale-pulse"
BALL_SCALE_RIPPLE = "ball-scale-ripple"
BALL_SCALE_RIPPLE_MULTIPLE = "ball-scale-ripple-multiple"
BALL_SPIN = "ball-spin"
BALL_SPIN_CLOCKWISE = "ball-spin-clockwise"
BALL_SPIN_CLOCKWISE_FADE = "ball-spin-clockwise-fade"
BALL_SPIN_CLOCKWISE_FADE_ROTATING = "ball-spin-clockwise-fade-rotating"
BALL_SPIN_FADE = "ball-spin-fade"
BALL_SPIN_FADE_ROTATING = "ball-spin-fade-rotating"
BALL_SPIN_ROTATE = "ball-spin-rotate"
BALL_SQUARE_CLOCKWISE_SPIN = "ball-square-clockwise-spin"
BALL_SQUARE_SPIN = "ball-square-spin"
BALL_TRIANGLE_PATH = "ball-triangle-path"
BALL_ZIG_ZAG = "ball-zig-zag"
BALL_ZIG_ZAG_DEFLECT = "ball-zig-zag-deflect"
COG = "cog"
CUBE_TRANSITION = "cube-transition"
FIRE = "fire"
LINE_SCALE = "line-scale"
LINE_SCALE_PARTY = "line-scale-party"
LINE_SCALE_PULSE_OUT = "line-scale-pulse-out"
LINE_SCALE_PULSE_OUT_RAPID = "line-scale-pulse-out-rapid"
LINE_SPIN_CLOCKWISE_FADE = "line-spin-clockwise-fade"
LINE_SPIN_CLOCKWISE_FADE_ROTATING = "line-spin-clockwise-fade-rotating"
LINE_SPIN_FADE = "line-spin-fade"
LINE_SPIN_FADE_ROTATING = "line-spin-fade-rotating"
PACMAN = "pacman"
SQUARE_JELLY_BOX = "square-jelly-box"
SQUARE_LOADER = "square-loader"
SQUARE_SPIN = "square-spin"
TIMER = "timer"
TRIANGLE_SKEW_SPIN = "triangle-skew-spin"
[docs]
class true_or_false(Enum):
true = "true"
false = "false"
[docs]
class PLACEHOLDER_VALUE(Enum):
PLACEHOLDER_VALUE = PLACEHOLDER
Port = Annotated[int, Field(gt=0, le=65535)]
NullablePort = Annotated[int, Field(ge=0, le=65535)]
PasswordScore = Annotated[int, Field(ge=0, le=4)]
GzipCompressionLevel = Annotated[int, Field(ge=1, le=9)]
NonNegativeInt = Annotated[int, Field(ge=0)]
AssignedCPU = Annotated[str, Field(pattern=r"^[0-9]+\.[0-9]+$")]
AssignedMemory = Annotated[str, Field(pattern=r"^[0-9]+(M|G)$")]
PostgresMem = Annotated[str, Field(pattern=r"^[0-9]+(KB|MB|GB)$")]
Neo4jMem = Annotated[str, Field(pattern="^[0-9]+(K|M|G|k|m|g)$")]
HealthcheckInterval = Annotated[str, Field(pattern=r"^[0-9]+(s|m|h)$")]
Version = Annotated[str, Field(pattern=r"^[0-9]+(\.[0-9]+)+$")]
# most of bool variables were deprecated since 1.0
# Backend and Frontend use different booleans due to Py vs Js
# 0/1 is a much more portable value to prevent true|True|"true"
# This fixes troubles in setting boolean values only used by Angular
# (expected true|false) or used by Pyton (expected True|False)
[docs]
class zero_or_one(IntEnum):
ZERO = 0
ONE = 1
[docs]
class ProjectModel(BaseModel):
title: str
description: str
keywords: str
rapydo: Version
version: str
[docs]
class SubmoduleModel(BaseModel):
online_url: str
branch: Optional[str] = None
_if: str
[docs]
class BaseEnvModel(BaseModel):
FRONTEND_FRAMEWORK: FRONTEND_FRAMEWORK_VALUES
FRONTEND_BUILD_MODE: FRONTEND_BUILD_MODE_VALUES
NETWORK_MTU: PositiveInt
DOCKER_LOGGING_DRIVER: DOCKER_LOGGING_DRIVERS
HEALTHCHECK_INTERVAL: HealthcheckInterval
HEALTHCHECK_BACKEND_CMD: str
LOG_LEVEL: LOG_LEVEL_VALUES
FILE_LOGLEVEL: LOG_LEVEL_VALUES
LOG_RETENTION: PositiveInt
MIN_PASSWORD_SCORE: PasswordScore
ACTIVATE_BACKEND: zero_or_one
ACTIVATE_PROXY: zero_or_one
ACTIVATE_ALCHEMY: zero_or_one
ACTIVATE_POSTGRES: zero_or_one
ACTIVATE_NEO4J: zero_or_one
ACTIVATE_RABBIT: zero_or_one
ACTIVATE_REDIS: zero_or_one
ACTIVATE_CELERY: zero_or_one
ACTIVATE_CELERYBEAT: zero_or_one
ACTIVATE_FLOWER: zero_or_one
ACTIVATE_FTP: zero_or_one
ACTIVATE_SMTP: zero_or_one
ACTIVATE_SMTP_SERVER: zero_or_one
ACTIVATE_SWAGGERUI: zero_or_one
ACTIVATE_ADMINER: zero_or_one
ACTIVATE_MYPY: zero_or_one
MYPY_DISALLOW_UNTYPED_DEFS: zero_or_one
MYPY_IGNORE_LIBS: str
MYPY_ADD_LIBS: str
MAX_LOGS_LENGTH: PositiveInt
APP_MODE: APP_MODE_VALUES
FLASK_HOST: str
FLASK_DEFAULT_PORT: Port
FLASK_DEBUG: zero_or_one
API_AUTOSTART: zero_or_one
BACKEND_PORT: Port
BACKEND_API_PORT: Port
BACKEND_URL: str
BACKEND_PYTHON_VERSION: BACKEND_PYTHON_VERSION_VALUES
PYTHON_MAIN_FILE: str
PYTHONASYNCIODEBUG: zero_or_one
PYTHONFAULTHANDLER: zero_or_one
PYTHONMALLOC: PYTHONMALLOC_VALUES
BACKEND_PREFIX: str
APP_SECRETS: Path
DATA_PATH: Path
DATA_IMPORT_FOLDER: Path
GUNICORN_WORKERS: PositiveInt
GUNICORN_WORKERS_PER_CORE: PositiveInt
GUNICORN_MAX_NUM_WORKERS: PositiveInt
CRONTAB_ENABLE: zero_or_one
GZIP_COMPRESSION_ENABLE: zero_or_one
GZIP_COMPRESSION_THRESHOLD: PositiveInt
GZIP_COMPRESSION_LEVEL: GzipCompressionLevel
ALEMBIC_AUTO_MIGRATE: zero_or_one
PROXY_HOST: str
PROXY_DEV_PORT: Port
PROXY_PROD_PORT: Port
PROXIED_CONNECTION: zero_or_one
DOMAIN_ALIASES: Optional[str] = None
SET_UNSAFE_EVAL: Optional[str] = None
SET_UNSAFE_INLINE: Optional[str] = None
SET_STYLE_UNSAFE_INLINE: Optional[str] = None
SET_CSP_SCRIPT_SRC: Optional[str] = None
SET_CSP_IMG_SRC: Optional[str] = None
SET_CSP_FONT_SRC: Optional[str] = None
SET_CSP_CONNECT_SRC: Optional[str] = None
SET_CSP_FRAME_SRC: Optional[str] = None
SET_MAX_REQUESTS_PER_SECOND_AUTH: PositiveInt
SET_MAX_REQUESTS_BURST_AUTH: PositiveInt
SET_MAX_REQUESTS_PER_SECOND_API: PositiveInt
SET_MAX_REQUESTS_BURST_API: PositiveInt
CORS_ALLOWED_ORIGIN: Optional[str] = None
SSL_VERIFY_CLIENT: zero_or_one
SSL_FORCE_SELF_SIGNED: zero_or_one
ALCHEMY_ENABLE_CONNECTOR: zero_or_one
ALCHEMY_EXPIRATION_TIME: PositiveInt
ALCHEMY_VERIFICATION_TIME: PositiveInt
ALCHEMY_HOST: str
ALCHEMY_PORT: Port
ALCHEMY_DBTYPE: ALCHEMY_DBTYPE_VALUES
ALCHEMY_USER: str
ALCHEMY_PASSWORD: str
ALCHEMY_DB: str
ALCHEMY_DBS: str
ALCHEMY_POOLSIZE: PositiveInt
POSTGRES_MAX_CONNECTIONS: PositiveInt
POSTGRES_SHARED_BUFFERS: PostgresMem
POSTGRES_WAL_BUFFERS: PostgresMem
POSTGRES_EFFECTIVE_CACHE_SIZE: PostgresMem
POSTGRES_WORK_MEM: PostgresMem
POSTGRES_MAINTENANCE_WORK_MEM: PostgresMem
POSTGRES_EFFECTIVE_IO_CONCURRENCY: PositiveInt
POSTGRES_MAX_WORKER_PROCESSES: PositiveInt
NEO4J_ENABLE_CONNECTOR: zero_or_one
NEO4J_EXPIRATION_TIME: PositiveInt
NEO4J_VERIFICATION_TIME: PositiveInt
NEO4J_HOST: str
NEO4J_BOLT_PORT: Port
NEO4J_USER: str
NEO4J_PASSWORD: str
NEO4J_EXPOSED_WEB_INTERFACE_PORT: Port
NEO4J_WEB_INTERFACE_PORT: Port
NEO4J_SSL_ENABLED: bool
NEO4J_BOLT_TLS_LEVEL: NEO4J_BOLT_TLS_LEVEL_VALUES
# They are equal to placeholder in production mode when neo4j is not enabled
NEO4J_HEAP_SIZE: Union[Neo4jMem, PLACEHOLDER_VALUE]
NEO4J_PAGECACHE_SIZE: Union[Neo4jMem, PLACEHOLDER_VALUE]
NEO4J_ALLOW_UPGRADE: bool
NEO4J_RECOVERY_MODE: bool
ELASTIC_HOST: str
ELASTIC_PORT: Port
RABBITMQ_ENABLE_CONNECTOR: zero_or_one
RABBITMQ_EXPIRATION_TIME: PositiveInt
RABBITMQ_VERIFICATION_TIME: PositiveInt
RABBITMQ_HOST: str
RABBITMQ_PORT: Port
RABBITMQ_VHOST: str
RABBITMQ_USER: str
RABBITMQ_PASSWORD: str
RABBITMQ_MANAGEMENT_PORT: Port
RABBITMQ_ENABLE_SHOVEL_PLUGIN: zero_or_one
RABBITMQ_SSL_CERTFILE: Optional[Path] = None
RABBITMQ_SSL_KEYFILE: Optional[Path] = None
RABBITMQ_SSL_FAIL_IF_NO_PEER_CERT: Optional[true_or_false] = None
RABBITMQ_SSL_ENABLED: zero_or_one
REDIS_ENABLE_CONNECTOR: zero_or_one
REDIS_EXPIRATION_TIME: PositiveInt
REDIS_VERIFICATION_TIME: PositiveInt
REDIS_HOST: str
REDIS_PORT: Port
REDIS_PASSWORD: str
FTP_ENABLE_CONNECTOR: zero_or_one
FTP_EXPIRATION_TIME: PositiveInt
FTP_VERIFICATION_TIME: PositiveInt
FTP_HOST: str
FTP_PORT: Port
FTP_USER: str
FTP_PASSWORD: str
FTP_SSL_ENABLED: zero_or_one
NFS_HOST: Optional[str] = None
NFS_EXPORTS_SECRETS: Path
NFS_EXPORTS_RABBITDATA: Path
NFS_EXPORTS_SQLDATA: Path
NFS_EXPORTS_GRAPHDATA: Path
NFS_EXPORTS_DATA_IMPORTS: Path
NFS_EXPORTS_PUREFTPD: Path
NFS_EXPORTS_SSL_CERTS: Path
NFS_EXPORTS_FLOWER_DB: Path
NFS_EXPORTS_REDISDATA: Path
CELERY_ENABLE_CONNECTOR: zero_or_one
CELERY_EXPIRATION_TIME: PositiveInt
CELERY_VERIFICATION_TIME: PositiveInt
CELERY_BROKER: CELERY_BROKER_VALUES
CELERY_BACKEND: CELERY_BACKEND_VALUES
CELERY_POOL_MODE: CELERY_POOL_MODE_VALUES
FLOWER_USER: str
FLOWER_PASSWORD: str
FLOWER_DBDIR: Path
FLOWER_PORT: Port
FLOWER_SSL_OPTIONS: Optional[str] = None
FLOWER_PROTOCOL: FLOWER_PROTOCOL_VALUES
DEFAULT_SCALE_BACKEND: PositiveInt
DEFAULT_SCALE_CELERY: PositiveInt
DEFAULT_SCALE_CELERYBEAT: PositiveInt
DEFAULT_SCALE_SWAGGERUI: PositiveInt
ASSIGNED_CPU_BACKEND: AssignedCPU
ASSIGNED_MEMORY_BACKEND: AssignedMemory
ASSIGNED_CPU_PROXY: AssignedCPU
ASSIGNED_MEMORY_PROXY: AssignedMemory
ASSIGNED_CPU_POSTGRES: AssignedCPU
ASSIGNED_MEMORY_POSTGRES: AssignedMemory
ASSIGNED_CPU_NEO4J: AssignedCPU
ASSIGNED_MEMORY_NEO4J: AssignedMemory
ASSIGNED_CPU_CELERY: AssignedCPU
ASSIGNED_MEMORY_CELERY: AssignedMemory
ASSIGNED_CPU_CELERYBEAT: AssignedCPU
ASSIGNED_MEMORY_CELERYBEAT: AssignedMemory
ASSIGNED_CPU_RABBIT: AssignedCPU
ASSIGNED_MEMORY_RABBIT: AssignedMemory
ASSIGNED_CPU_REDIS: AssignedCPU
ASSIGNED_MEMORY_REDIS: AssignedMemory
ASSIGNED_CPU_FLOWER: AssignedCPU
ASSIGNED_MEMORY_FLOWER: AssignedMemory
ASSIGNED_CPU_SWAGGERUI: AssignedCPU
ASSIGNED_MEMORY_SWAGGERUI: AssignedMemory
ASSIGNED_CPU_ADMINER: AssignedCPU
ASSIGNED_MEMORY_ADMINER: AssignedMemory
ASSIGNED_CPU_FTP: AssignedCPU
ASSIGNED_MEMORY_FTP: AssignedMemory
ASSIGNED_CPU_SMTP: AssignedCPU
ASSIGNED_MEMORY_SMTP: AssignedMemory
ASSIGNED_CPU_REGISTRY: AssignedCPU
ASSIGNED_MEMORY_REGISTRY: AssignedMemory
REGISTRY_HOST: Optional[str] = None
REGISTRY_PORT: Port
REGISTRY_USERNAME: str
REGISTRY_PASSWORD: str
REGISTRY_HTTP_SECRET: Optional[str] = None
ACTIVATE_FAIL2BAN: zero_or_one
SWARM_MANAGER_ADDRESS: Optional[str] = None
# SYSLOG_ADDRESS: Optional[str]
SMTP_ENABLE_CONNECTOR: zero_or_one
SMTP_EXPIRATION_TIME: PositiveInt
SMTP_VERIFICATION_TIME: PositiveInt
SMTP_ADMIN: Optional[str] = None
SMTP_NOREPLY: Optional[str] = None
SMTP_REPLYTO: Optional[str] = None
SMTP_HOST: Optional[str] = None
SMTP_PORT: NullablePort
SMTP_USERNAME: Optional[str] = None
SMTP_PASSWORD: Optional[str] = None
SMTP_SERVER_HOST: str
SMTP_SERVER_PORT: Port
FRONTEND_URL: str
FRONTEND_PREFIX: str
ALLOW_PASSWORD_RESET: zero_or_one
ALLOW_REGISTRATION: zero_or_one
ALLOW_TERMS_OF_USE: zero_or_one
REGISTRATION_NOTIFICATIONS: zero_or_one
SENTRY_URL: Optional[str] = None
SHOW_LOGIN: zero_or_one
ENABLE_FOOTER: zero_or_one
ENABLE_ANGULAR_SSR: zero_or_one
ENABLE_YARN_PNP: zero_or_one
ENABLE_ANGULAR_MULTI_LANGUAGE: zero_or_one
FORCE_SSR_SERVER_MODE: zero_or_one
SPINNER_TYPE: SPINNER_TYPES
ACTIVATE_AUTH: zero_or_one
AUTH_SERVICE: AUTH_SERVICE_VALUES
AUTH_DEFAULT_USERNAME: str
AUTH_DEFAULT_PASSWORD: str
AUTH_MIN_PASSWORD_LENGTH: NonNegativeInt
AUTH_FORCE_FIRST_PASSWORD_CHANGE: zero_or_one
AUTH_MAX_PASSWORD_VALIDITY: NonNegativeInt
AUTH_DISABLE_UNUSED_CREDENTIALS_AFTER: NonNegativeInt
AUTH_MAX_LOGIN_ATTEMPTS: NonNegativeInt
AUTH_LOGIN_BAN_TIME: PositiveInt
AUTH_SECOND_FACTOR_AUTHENTICATION: zero_or_one
AUTH_TOTP_VALIDITY_WINDOW: NonNegativeInt
AUTH_JWT_TOKEN_TTL: PositiveInt
AUTH_TOKEN_SAVE_FREQUENCY: NonNegativeInt
AUTH_TOKEN_IP_GRACE_PERIOD: NonNegativeInt
ALLOW_ACCESS_TOKEN_PARAMETER: zero_or_one
DEFAULT_DHLEN: PositiveInt
PASSWORD_EXPIRATION_WARNING: NonNegativeInt
FORCE_PRODUCTION_TESTS: zero_or_one
[docs]
class CoreEnvModel(BaseEnvModel):
model_config = ConfigDict(extra="forbid")
[docs]
class CustomEnvModel(BaseEnvModel):
model_config = ConfigDict(extra="ignore")
[docs]
class CoreVariablesModel(BaseModel):
submodules: dict[str, SubmoduleModel]
roles: dict[str, str]
env: CoreEnvModel
[docs]
class CustomVariablesModel(BaseModel):
submodules: dict[str, SubmoduleModel]
roles: dict[str, str]
env: CustomEnvModel
[docs]
class CoreConfigurationModel(BaseModel):
project: ProjectModel
tags: dict[str, str]
variables: CoreVariablesModel
[docs]
class CustomConfigurationModel(BaseModel):
project: ProjectModel
tags: dict[str, str]
variables: CustomVariablesModel
[docs]
def read_configuration(
default_file_path: Path,
base_project_path: Path,
projects_path: Path,
submodules_path: Path,
read_extended: bool = True,
production: bool = False,
) -> tuple[Configuration, Optional[str], Optional[Path], Configuration]:
"""
Read default configuration
"""
custom_configuration = load_yaml_file(
file=base_project_path.joinpath(PROJECT_CONF_FILENAME)
)
# Verify custom project configuration
project = custom_configuration.get("project")
# Can't be tested because it is included in default configuration
if project is None: # pragma: no cover
raise AttributeError("Missing project configuration")
variables = ["title", "description", "version", "rapydo"]
for key in variables:
# Can't be tested because it is included in default configuration
if project.get(key) is None: # pragma: no cover
print_and_exit(
"Project not configured, missing key '{}' in file {}/{}",
key,
base_project_path,
PROJECT_CONF_FILENAME,
)
base_configuration = load_yaml_file(
file=default_file_path.joinpath(PROJECTS_DEFAULTS_FILE)
)
# Prevent any change due to the mix_configuration
base_configuration_copy = deepcopy(base_configuration)
if production:
base_prod_conf = load_yaml_file(
file=default_file_path.joinpath(PROJECTS_PROD_DEFAULTS_FILE)
)
base_configuration = mix_configuration(base_configuration, base_prod_conf)
if read_extended:
extended_project = project.get("extends")
else:
extended_project = None
if extended_project is None:
# Mix default and custom configuration
return (
mix_configuration(base_configuration, custom_configuration),
None,
None,
base_configuration_copy,
)
extends_from = project.get("extends_from") or "projects"
if extends_from == "projects":
extend_path = projects_path.joinpath(extended_project)
elif extends_from.startswith("submodules/"): # pragma: no cover
repository_name = (extends_from.split("/")[1]).strip()
if repository_name == "":
print_and_exit("Invalid repository name in extends_from, name is empty")
extend_path = submodules_path.joinpath(
repository_name, projects_path, extended_project
)
else: # pragma: no cover
suggest = "Expected values: 'projects' or 'submodules/${REPOSITORY_NAME}'"
print_and_exit("Invalid extends_from parameter: {}.\n{}", extends_from, suggest)
if not extend_path.exists(): # pragma: no cover
print_and_exit("From project not found: {}", extend_path)
extended_configuration = load_yaml_file(
file=extend_path.joinpath(PROJECT_CONF_FILENAME)
)
m1 = mix_configuration(base_configuration, extended_configuration)
return (
mix_configuration(m1, custom_configuration),
extended_project,
Path(extend_path),
base_configuration_copy,
)
# This function is not mypy-friend after the introduction of TypedDict
# TypedDict key must be a string literal;
# This use case is not supported by mypy
# https://github.com/python/mypy/issues/7178
[docs]
def mix_configuration(
base: Optional[Configuration], custom: Optional[Configuration]
) -> Configuration:
# WARNING: This function has the side effect of changing the input base dict!
if base is None:
base = {}
if custom is None:
return base
for key, elements in custom.items():
if key not in base:
# TypedDict key must be a string literal;
base[key] = custom[key] # type: ignore
continue
if elements is None: # pragma: no cover
# TypedDict key must be a string literal;
if isinstance(base[key], dict): # type: ignore
log.warning("Cannot replace {} with empty list", key)
continue
if isinstance(elements, dict):
# TypedDict key must be a string literal;
base[key] = mix_configuration(base[key], custom[key]) # type: ignore
elif isinstance(elements, list):
for e in elements: # pragma: no cover
# TypedDict key must be a string literal;
base[key].append(e) # type: ignore
else:
# TypedDict key must be a string literal;
base[key] = elements # type: ignore
return base
[docs]
def load_yaml_file(
file: Path, keep_order: bool = False, is_optional: bool = False
) -> Configuration:
"""
Import any data from a YAML file.
"""
if not file.exists():
if not is_optional:
print_and_exit("Failed to read {}: File does not exist", file)
return {}
with open(file) as fh:
try:
docs = list(yaml.safe_load_all(fh))
if not docs:
print_and_exit("YAML file is empty: {}", file)
# Return value of yaml.safe_load_all is un-annotated and considered as Any
# But we known that it is a Dict Configuration-compliant
return cast(Configuration, docs[0])
except Exception as e:
# # IF dealing with a strange exception string (escaped)
# import codecs
# error, _ = codecs.getdecoder("unicode_escape")(str(error))
print_and_exit("Failed to read [{}]: {}", file, str(e))
[docs]
def read_composer_yamls(config_files: list[Path]) -> tuple[list[Path], list[Path]]:
base_files: list[Path] = []
all_files: list[Path] = []
# YAML CHECK UP
for path in config_files:
try:
# This is to verify that mandatory files exist and yml syntax is valid
conf = load_yaml_file(file=path, is_optional=False)
if conf.get("version") != COMPOSE_FILE_VERSION: # pragma: no cover
log.warning(
"Compose file version in {} is {}, expected {}",
path,
conf.get("version"),
COMPOSE_FILE_VERSION,
)
if path.exists():
all_files.append(path)
# Base files are those loaded from CONFS_DIR
if CONFS_DIR in path.parents:
base_files.append(path)
except KeyError as e: # pragma: no cover
print_and_exit("Error reading {}: {}", path, str(e))
return all_files, base_files
[docs]
def validate_configuration(conf: Configuration, core: bool) -> None:
if conf:
try:
if core:
CoreConfigurationModel(**conf)
else:
CustomConfigurationModel(**conf)
except ValidationError as e:
for field in str(e).split("\n")[1::2]:
# field is like:
# "variables -> env -> XYZ"
# this way it is converted in key = variables.env.XYZ
key = ".".join(field.split(" -> "))
log.error(
"Invalid value for {}: {}", field, glom(conf, key, default=None)
)
print_and_exit(str(e))
[docs]
def validate_env(env: dict[str, EnvType]) -> None:
try:
BaseEnvModel(**env)
except ValidationError as e:
for field in str(e).split("\n")[1::2]:
log.error("Invalid value for {}: {}", field, env.get(field, "N/A"))
print_and_exit(str(e))