Rebuild configuration loading mechanisms with Flask tools

- and extract Flask app initialization mostly into module
master
Felix Stupp 1 year ago
parent ffc0550342
commit cf66f47b2e
Signed by: zocker
GPG Key ID: 93E1BD26F6B02FB7

@ -5,12 +5,9 @@
from __future__ import annotations
from datetime import datetime, timedelta
from functools import partial
import io
import itertools
import logging
import os
from pathlib import Path
import re
from urllib.parse import urlencode, quote_plus
from typing import (
@ -28,7 +25,6 @@ from typing import (
)
from flask import (
Flask,
make_response,
redirect,
request,
@ -44,6 +40,7 @@ from pony import orm
from entertainment_decider import (
common,
create_app,
)
from entertainment_decider.models import (
MediaCollection,
@ -96,25 +93,7 @@ logging.basicConfig(format="%(asctime)s === %(message)s", level=logging.DEBUG)
####
DEBUG_DATABASE = False
flask_app = Flask(
__name__,
static_folder=str(Path(__file__).parent / "static"),
)
flask_app.config.update(
dict(
CELERY=dict(),
DEBUG=True,
PONY=dict(
provider="sqlite",
filename="./db.sqlite",
create_db=True,
)
if DEBUG_DATABASE
else dict(),
)
)
flask_app = create_app()
def environ_bool(value: Union[str, bool]) -> bool:
@ -158,93 +137,12 @@ def environ_timedelta_seconds(value: Union[int, str, timedelta]) -> int:
return environ_timedelta(value) // timedelta(seconds=1)
ConfigKeySetter = Callable[[str, Any], None]
ConfigSingleTranslator = Callable[[Any], Any]
ConfigTranslatorIterable = Iterable[ConfigSingleTranslator]
ConfigTranslatorCreator = Callable[[str], ConfigTranslatorIterable]
def config_suffixer(
setter: ConfigKeySetter,
prefix: str,
lower: bool = True,
) -> ConfigTranslatorCreator:
def creator(key: str) -> ConfigTranslatorIterable:
if not key.startswith(prefix):
raise Exception(f"Environment key {key!r} is missing suffix {prefix!r}")
new_key = key[len(prefix) :]
new_key = new_key.lower() if lower else new_key
return (partial(setter, new_key),)
return creator
def celery_config_setter(key: str, val: Any) -> None:
flask_app.config["CELERY"][key] = val
celery_config_same = config_suffixer(celery_config_setter, "CELERY_")
def flask_config_setter(key: str, val: Any) -> None:
flask_app.config[key] = val
flask_config_same = config_suffixer(flask_config_setter, "FLASK_", lower=False)
def pony_config_setter(key: str, val: Any) -> None:
flask_app.config["PONY"][key] = val
pony_config_same = config_suffixer(pony_config_setter, "PONY_")
CONFIG_TRANSLATE_TABLE: Dict[
str, Union[ConfigTranslatorIterable, ConfigTranslatorCreator]
] = {
"CELERY_BROKER_URL": celery_config_same,
"CELERY_RESULT_BACKEND": celery_config_same,
"FLASK_DEBUG": (
environ_bool,
partial(flask_config_setter, "DEBUG"),
),
"PONY_PROVIDER": pony_config_same,
"PONY_FILENAME": pony_config_same,
"PONY_CREATE_DB": (
environ_bool,
partial(pony_config_setter, "create_db"),
),
"PONY_HOST": pony_config_same,
"PONY_PORT": (
environ_int,
partial(pony_config_setter, "port"),
),
"PONY_DATABASE": pony_config_same,
"PONY_DB": pony_config_same,
"PONY_USER": pony_config_same,
"PONY_PASSWORD": pony_config_same,
"PONY_PASSWD": pony_config_same,
"PONY_DSN": pony_config_same,
"PONY_CHARSET": pony_config_same,
}
for key, val in os.environ.items():
trans = CONFIG_TRANSLATE_TABLE.get(key)
if trans is not None:
trans = trans(key) if callable(trans) else trans
res: Any = val
for caller in trans:
new_res = caller(res)
if new_res is not None:
res = new_res
####
# Pony init
####
db.bind(**flask_app.config["PONY"])
db.bind(**flask_app.config.get_namespace("PONY_"))
db.generate_mapping(create_tables=True)
setup_custom_tables()

@ -0,0 +1,25 @@
from __future__ import annotations
from pathlib import Path
from flask import (
Flask,
)
from .web.config import apply_config
def create_app() -> Flask:
app = Flask(
import_name=__name__,
instance_relative_config=True,
root_path=str(Path(__file__).parent.parent),
)
apply_config(app.config)
print(app.static_folder)
return app
__all__ = [
"create_app",
]

@ -0,0 +1,16 @@
from .load import load_config
from .validate import validate_config
from flask import (
Config,
)
def apply_config(config: Config) -> None:
load_config(config)
validate_config(config)
__all__ = [
"apply_config",
]

@ -0,0 +1,26 @@
from __future__ import annotations
class _Config:
pass
class DebugConfig(_Config):
PONY_CREATE_DB = True
PONY_FILENAME = "./db.sqlite"
PONY_PROVIDER = "sqlite"
class TestingConfig(_Config):
pass
class ProductionConfig(_Config):
pass
__all__ = [
"DebugConfig",
"ProductionConfig",
"TestingConfig",
]

@ -0,0 +1,48 @@
from __future__ import annotations
from flask import (
Config,
)
from .defaults import (
DebugConfig,
ProductionConfig,
TestingConfig,
)
FLASK_DEBUG_KEY = "DEBUG"
FLASK_TESTING_KEY = "TESTING"
ENV_FLASK_PREFIX = "FLASK"
ENV_OTHER_NAMESPACES = [
"CELERY",
"PONY",
]
def load_config(config: Config) -> None:
_load_defaults(config)
_load_from_env(config)
def _load_defaults(config: Config) -> None:
defaults = (
DebugConfig
if config.get(FLASK_DEBUG_KEY, False)
else TestingConfig
if config.get(FLASK_TESTING_KEY, False)
else ProductionConfig
)
config.from_object(defaults())
def _load_from_env(config: Config) -> None:
config.from_prefixed_env(
prefix="FLASK",
)
for prefix in ENV_OTHER_NAMESPACES:
config.from_prefixed_env(
prefix=prefix,
trim_prefix=False,
)

@ -0,0 +1,106 @@
from __future__ import annotations
from typing import (
Mapping,
Optional,
Tuple,
Type,
TypeAlias,
)
# type aliases
TypeSet: TypeAlias = Tuple[Type, ...]
TypeLike: TypeAlias = Type | TypeSet
# exceptions
class ConfigValidationError(Exception):
key: str
class ConfigMissingRequiredError(ConfigValidationError):
def __init__(self, *, key: str) -> None:
super().__init__(f"Missing configuration value for key {key!r}")
self.key = key
class ConfigInvalidTypeError(ConfigValidationError):
key: str
types: TypeSet
def __init__(self, *, key: str, types: TypeLike, got_type: Type) -> None:
self.key = key
self.types = self.__parse_type_like(types)
super().__init__(
f"Required configuration value for key {key!r} to be of either {self.type_list}; got type {got_type}."
)
@staticmethod
def __parse_type_like(types: TypeLike) -> TypeSet:
if isinstance(types, tuple):
return types
return (types,)
@property
def type_list(self) -> str:
return ", ".join(str(t) for t in self.types)
# methods
_REQUIRED_KEYS: Mapping[str, Optional[TypeLike]] = {
"PONY_PROVIDER": str,
"SECRET_KEY": (bytes, str),
}
_OPTIONAL_KEYS: Mapping[str, Optional[TypeLike]] = {
"PONY_CHARSET": str,
"PONY_CREATE_DB": bool,
"PONY_DATABASE": str,
"PONY_DB": str,
"PONY_DSN": str,
"PONY_FILENAME": str,
"PONY_HOST": str,
"PONY_PASSWD": str,
"PONY_PASSWORD": str,
"PONY_PORT": int,
"PONY_USER": str,
}
def validate_config(config: Mapping) -> None:
for required, key_list in {True: _REQUIRED_KEYS, False: _OPTIONAL_KEYS}.items():
for key, types in key_list.items():
check_value(
config=config,
key=key,
types=types,
required=required,
)
def check_value(
config: Mapping, key: str, types: Optional[TypeLike] = None, required: bool = True
) -> None:
if key not in config:
if required:
raise ConfigMissingRequiredError(
key=key,
)
return
if types is None:
return
value = config[key]
if not isinstance(value, types):
raise ConfigInvalidTypeError(
key=key,
types=types,
got_type=type(value),
)
Loading…
Cancel
Save