diff --git a/CHANGES.md b/CHANGES.md index 11d8854..f7847c1 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,4 +1,6 @@ # Changes +## 0.8.0 (2020-03-31) +- `[feature]` Settings plugin ## 0.7.0 (2020-03-29) - `[feature]` Control plugin with Health, Heartbeat, Environment and Version ## 0.6.1 (2020-03-24) diff --git a/README.md b/README.md index c5a3183..fb9ddf9 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,9 @@

# fastapi-plugins -FastAPI framework plugins +FastAPI framework plugins - simple way to share `fastapi` code and utilities across applications. + +The concept is `plugin` - plug a functional utility into your application without or with minimal effort. * [Cache](./docs/cache.md) * [Memcached](./docs/cache.md#memcached) @@ -28,6 +30,7 @@ FastAPI framework plugins * [Environment](./docs/control.md#environment) * [Health](./docs/control.md#health) * [Heartbeat](./docs/control.md#heartbeat) +* [Application settings/configuration](./docs/settings.md) * Celery * MQ * and much more is already in progress... @@ -50,6 +53,13 @@ pip install fastapi-plugins[all] ``` ## Quick start +### Plugin +Add information about plugin system. +### Application settings +Add information about settings. +### Application configuration +Add information about configuration of an application +### Complete example ```python import fastapi import fastapi_plugins @@ -63,6 +73,7 @@ import asyncio import aiojobs import aioredis +@fastapi_plugins.registered_configuration class AppSettings( fastapi_plugins.ControlSettings, fastapi_plugins.RedisSettings, @@ -71,14 +82,22 @@ class AppSettings( ): api_name: str = str(__name__) + +@fastapi_plugins.registered_configuration(name='sentinel') +class AppSettingsSentinel(AppSettings): + redis_type = fastapi_plugins.RedisType.sentinel + redis_sentinels = 'localhost:26379' + + app = fastapi.FastAPI() -config = AppSettings() +config = fastapi_plugins.get_config() @app.get("/") async def root_get( cache: aioredis.Redis=fastapi.Depends(fastapi_plugins.depends_redis), + conf: pydantic.BaseSettings=fastapi.Depends(fastapi_plugins.depends_config) # noqa E501 ) -> typing.Dict: - return dict(ping=await cache.ping()) + return dict(ping=await cache.ping(), api_name=conf.api_name) @app.post("/jobs/schedule/") @@ -132,6 +151,8 @@ async def memcached_demo_post( @app.on_event('startup') async def on_startup() -> None: + await fastapi_plugins.config_plugin.init_app(app, config) + await fastapi_plugins.config_plugin.init() await memcached_plugin.init_app(app, config) await memcached_plugin.init() await fastapi_plugins.redis_plugin.init_app(app, config=config) @@ -153,6 +174,7 @@ async def on_shutdown() -> None: await fastapi_plugins.scheduler_plugin.terminate() await fastapi_plugins.redis_plugin.terminate() await memcached_plugin.terminate() + await fastapi_plugins.config_plugin.terminate() ``` # Development diff --git a/docs/settings.md b/docs/settings.md new file mode 100644 index 0000000..1705f3b --- /dev/null +++ b/docs/settings.md @@ -0,0 +1,163 @@ +# Settings +The easy way to configure the `FastAPI` application: +* define (various) configuration(s) +* register configuration +* use configuration depending on the environment +* use configuration in router handler if required + +## Define configuration +It is a good practice to defined various configuration for your application. +```python + import fastapi_plugins + + class DefaultSettings(fastapi_plugins.RedisSettings): + api_name: str = str(__name__) + + class DockerSettings(DefaultSettings): + redis_type = fastapi_plugins.RedisType.sentinel + redis_sentinels = 'localhost:26379' + + class LocalSettings(DefaultSettings): + pass + + class TestSettings(DefaultSettings): + testing: bool = True + + class MyConfigSettings(DefaultSettings): + custom: bool = True + + ... +``` + +## Register configuration +Registration with a decorator +```python + import fastapi_plugins + + class DefaultSettings(fastapi_plugins.RedisSettings): + api_name: str = str(__name__) + + # @fastapi_plugins.registered_configuration_docker + @fastapi_plugins.registered_configuration + class DockerSettings(DefaultSettings): + redis_type = fastapi_plugins.RedisType.sentinel + redis_sentinels = 'localhost:26379' + + @fastapi_plugins.registered_configuration_local + class LocalSettings(DefaultSettings): + pass + + @fastapi_plugins.registered_configuration_test + class TestSettings(DefaultSettings): + testing: bool = True + + @fastapi_plugins.registered_configuration(name='my_config') + class MyConfigSettings(DefaultSettings): + custom: bool = True + ... +``` + +or by a function call +```python + import fastapi_plugins + + class DefaultSettings(fastapi_plugins.RedisSettings): + api_name: str = str(__name__) + + class DockerSettings(DefaultSettings): + redis_type = fastapi_plugins.RedisType.sentinel + redis_sentinels = 'localhost:26379' + + class LocalSettings(DefaultSettings): + pass + + class TestSettings(DefaultSettings): + testing: bool = True + + class MyConfigSettings(DefaultSettings): + custom: bool = True + + fastapi_plugins.register_config(DockerSettings) + # fastapi_plugins.register_config_docker(DockerSettings) + fastapi_plugins.register_config_local(LocalSettings) + fastapi_plugins.register_config_test(TestSettings) + fastapi_plugins.register_config(MyConfigSettings, 'my_config') + ... +``` + + +## Application configuration +Next, create application and it's configuration, and register the plugin if needed. The last is optinally, +and is only relevant for use cases, where the configuration values are required for endpoint handlers (see `api_name` below). +```python +import fastapi +import fastapi_plugins +... + +app = fastapi.FastAPI() +config = fastapi_plugins.get_config() + +@app.get("/") +async def root_get( + cache: aioredis.Redis=fastapi.Depends(fastapi_plugins.depends_redis), + conf: pydantic.BaseSettings=fastapi.Depends(fastapi_plugins.depends_config) # noqa E501 +) -> typing.Dict: + return dict(ping=await cache.ping(), api_name=conf.api_name) + +@app.on_event('startup') +async def on_startup() -> None: + await fastapi_plugins.config_plugin.init_app(app, config) + await fastapi_plugins.config_plugin.init() + await fastapi_plugins.redis_plugin.init_app(app, config=config) + await fastapi_plugins.redis_plugin.init() + await fastapi_plugins.control_plugin.init_app(app, config=config, version=__version__, environ=config.dict()) + await fastapi_plugins.control_plugin.init() + + +@app.on_event('shutdown') +async def on_shutdown() -> None: + await fastapi_plugins.control_plugin.terminate() + await fastapi_plugins.redis_plugin.terminate() + await fastapi_plugins.config_plugin.terminate() +``` + +## Use configuration +Now, the application will use by default a configuration for `docker` or by user's command any other configuration. +```bash + uvicorn scripts.demo_app:app + curl -X 'GET' 'http://localhost:8000/control/environ' -H 'accept: application/json' + { + "environ": { + "config_name":"docker", + "redis_type":"sentinel", + ... + } + } + + ... + CONFIG_NAME=local uvicorn scripts.demo_app:app + curl -X 'GET' 'http://localhost:8000/control/environ' -H 'accept: application/json' + { + "environ": { + "config_name":"local", + "redis_type":"redis", + ... + } + } +``` + +It is also usefull with `docker-compose`: +```yaml +services: + demo_fastapi_plugin: + image: demo_fastapi_plugin + environment: + - CONFIG_NAME=docker + + ... + + demo_fastapi_plugin: + image: demo_fastapi_plugin + environment: + - CONFIG_NAME=docker_sentinel +``` diff --git a/fastapi_plugins/__init__.py b/fastapi_plugins/__init__.py index 21b05ad..1b91933 100644 --- a/fastapi_plugins/__init__.py +++ b/fastapi_plugins/__init__.py @@ -17,6 +17,7 @@ from .control import * # noqa F401 F403 from ._redis import * # noqa F401 F403 from .scheduler import * # noqa F401 F403 +from .settings import * # noqa F401 F403 from .version import VERSION # try: @@ -41,8 +42,6 @@ # TODO: provide a generic cache type (redis, memcached, in-memory) # and share some settings. Module/Sub-Pack cache -# TODO: health - # TODO: databases # TODO: mq - activemq, rabbitmq, kafka @@ -51,8 +50,6 @@ # TODO: celery -# TOOD: abstract routers with configurable endpoints - # TODO: check socketio (python-socketio) - do we need this? # TODO: look at fastapi-cache (memcache?) look at mqtt? diff --git a/fastapi_plugins/_redis.py b/fastapi_plugins/_redis.py index ec6d0a3..a3da486 100644 --- a/fastapi_plugins/_redis.py +++ b/fastapi_plugins/_redis.py @@ -192,5 +192,7 @@ async def health(self) -> typing.Dict: redis_plugin = RedisPlugin() -async def depends_redis(request: starlette.requests.Request) -> aioredis.Redis: - return await request.app.state.REDIS() +async def depends_redis( + conn: starlette.requests.HTTPConnection +) -> aioredis.Redis: + return await conn.app.state.REDIS() diff --git a/fastapi_plugins/control.py b/fastapi_plugins/control.py index 3498470..1a28c8e 100644 --- a/fastapi_plugins/control.py +++ b/fastapi_plugins/control.py @@ -128,6 +128,15 @@ class ControlHeartBeat(ControlBaseModel): ) +# class ControlInfo(ControlBaseModel): +# status: str = pydantic.Field( +# 'API is up and running', +# title='status', +# min_length=1, +# exampe='API is up and running' +# ) + + class ControlVersion(ControlBaseModel): version: str = pydantic.Field( ..., @@ -322,7 +331,7 @@ async def init_app( raise ControlError('Control configuration is not initialized') elif not isinstance(self.config, self.DEFAULT_CONFIG_CLASS): raise ControlError('Control configuration is not valid') - app.state.CONTROL = self + app.state.PLUGIN_CONTROL = self # # initialize here while `app` is available self.controller = Controller( @@ -361,6 +370,6 @@ async def terminate(self): async def depends_control( - request: starlette.requests.Request + conn: starlette.requests.HTTPConnection ) -> Controller: - return await request.app.state.CONTROL() + return await conn.app.state.PLUGIN_CONTROL() diff --git a/fastapi_plugins/memcached.py b/fastapi_plugins/memcached.py index d07f060..820d4b0 100644 --- a/fastapi_plugins/memcached.py +++ b/fastapi_plugins/memcached.py @@ -132,6 +132,6 @@ async def health(self) -> typing.Dict: async def depends_memcached( - request: starlette.requests.Request + conn: starlette.requests.HTTPConnection ) -> MemcachedClient: - return await request.app.state.MEMCACHED() + return await conn.app.state.MEMCACHED() diff --git a/fastapi_plugins/scheduler.py b/fastapi_plugins/scheduler.py index 626f71d..040b05c 100644 --- a/fastapi_plugins/scheduler.py +++ b/fastapi_plugins/scheduler.py @@ -169,6 +169,6 @@ async def health(self) -> typing.Dict: async def depends_scheduler( - request: starlette.requests.Request + conn: starlette.requests.HTTPConnection ) -> aiojobs.Scheduler: - return await request.app.state.AIOJOBS_SCHEDULER() + return await conn.app.state.AIOJOBS_SCHEDULER() diff --git a/fastapi_plugins/settings.py b/fastapi_plugins/settings.py new file mode 100644 index 0000000..78dd53d --- /dev/null +++ b/fastapi_plugins/settings.py @@ -0,0 +1,190 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# fastapi_plugins.settings +''' +:author: madkote +:contact: madkote(at)bluewin.ch +:copyright: Copyright 2021, madkote RES + +fastapi_plugins.settings +------------------------ +Settings plugin +''' + +from __future__ import absolute_import + +import functools +import typing + +import fastapi +import pydantic +import starlette.config + +from .plugin import PluginError +from .plugin import Plugin +from .version import VERSION + +__all__ = [ + 'ConfigError', 'ConfigPlugin', 'depends_config', 'config_plugin', + # + 'register_config', + 'register_config_docker', 'register_config_local', 'register_config_test', + # 'register_config_by_name', + 'reset_config', 'get_config', + # + 'registered_configuration', 'registered_configuration_docker', + 'registered_configuration_local', 'registered_configuration_test', + # 'registered_configuration_by_name', + # + 'DEFAULT_CONFIG_ENVVAR', 'DEFAULT_CONFIG_NAME', + 'CONFIG_NAME_DEFAULT', 'CONFIG_NAME_DOCKER', 'CONFIG_NAME_LOCAL', + 'CONFIG_NAME_TEST', +] +__author__ = 'madkote ' +__version__ = '.'.join(str(x) for x in VERSION) +__copyright__ = 'Copyright 2021, madkote RES' + +DEFAULT_CONFIG_ENVVAR: str = 'CONFIG_NAME' +DEFAULT_CONFIG_NAME: str = 'docker' + +CONFIG_NAME_DEFAULT = DEFAULT_CONFIG_NAME +CONFIG_NAME_DOCKER = DEFAULT_CONFIG_NAME +CONFIG_NAME_LOCAL = 'local' +CONFIG_NAME_TEST = 'test' + + +class ConfigError(PluginError): + pass + + +class ConfigManager(object): + def __init__(self): + self._settings_map = {} + + def register(self, name: str, config: pydantic.BaseSettings) -> None: + self._settings_map[name] = config + + def reset(self) -> None: + self._settings_map.clear() + + def get_config( + self, + config_or_name: typing.Union[str, pydantic.BaseSettings]=None, + config_name_default: str=DEFAULT_CONFIG_NAME, + config_name_envvar: str=DEFAULT_CONFIG_ENVVAR + ) -> pydantic.BaseSettings: + if isinstance(config_or_name, pydantic.BaseSettings): + return config_or_name + if not config_or_name: + config_or_name = config_name_default + base_cfg = starlette.config.Config() + config_or_name = base_cfg( + config_name_envvar, + cast=str, + default=config_or_name + ) + if config_or_name not in self._settings_map: + raise ConfigError('Unknown configuration "%s"' % config_or_name) + return self._settings_map[config_or_name]() + + +_manager = ConfigManager() + + +def register_config(config: pydantic.BaseSettings, name: str=None) -> None: + if not name: + name = CONFIG_NAME_DEFAULT + _manager.register(name, config) + + +def register_config_docker(config: pydantic.BaseSettings) -> None: + _manager.register(CONFIG_NAME_DOCKER, config) + + +def register_config_local(config: pydantic.BaseSettings) -> None: + _manager.register(CONFIG_NAME_LOCAL, config) + + +def register_config_test(config: pydantic.BaseSettings) -> None: + _manager.register(CONFIG_NAME_TEST, config) + + +def registered_configuration(cls=None, /, *, name: str=None): + if not name: + name = CONFIG_NAME_DEFAULT + + def wrap(kls): + _manager.register(name, kls) + return kls + + if cls is None: + return wrap + return wrap(cls) + + +def registered_configuration_docker(cls=None): + def wrap(kls): + _manager.register(CONFIG_NAME_DOCKER, kls) + return kls + if cls is None: + return wrap + return wrap(cls) + + +def registered_configuration_local(cls=None): + def wrap(kls): + _manager.register(CONFIG_NAME_LOCAL, kls) + return kls + if cls is None: + return wrap + return wrap(cls) + + +def registered_configuration_test(cls=None): + def wrap(kls): + _manager.register(CONFIG_NAME_TEST, kls) + return kls + if cls is None: + return wrap + return wrap(cls) + + +def reset_config() -> None: + _manager.reset() + + +@functools.lru_cache() +def get_config( + config_or_name: typing.Union[str, pydantic.BaseSettings]=None, + config_name_default: str=DEFAULT_CONFIG_NAME, + config_name_envvar: str=DEFAULT_CONFIG_ENVVAR +) -> pydantic.BaseSettings: + return _manager.get_config( + config_or_name=config_or_name, + config_name_default=config_name_default, + config_name_envvar=config_name_envvar + ) + + +class ConfigPlugin(Plugin): + DEFAULT_CONFIG_CLASS = pydantic.BaseSettings + + async def _on_call(self) -> pydantic.BaseSettings: + return self.config + + async def init_app( + self, + app: fastapi.FastAPI, + config: pydantic.BaseSettings=None, + ) -> None: + self.config = config or self.DEFAULT_CONFIG_CLASS() + app.state.PLUGIN_CONFIG = self + + +config_plugin = ConfigPlugin() + + +async def depends_config( + conn: starlette.requests.HTTPConnection +) -> pydantic.BaseSettings: + return await conn.app.state.PLUGIN_CONFIG() diff --git a/fastapi_plugins/version.py b/fastapi_plugins/version.py index 3f86773..8ff6d30 100644 --- a/fastapi_plugins/version.py +++ b/fastapi_plugins/version.py @@ -13,7 +13,7 @@ from __future__ import absolute_import -VERSION = (0, 7, 0) +VERSION = (0, 8, 0) __all__ = [] __author__ = 'madkote ' diff --git a/scripts/demo_app.py b/scripts/demo_app.py index 7672f8c..232acf3 100644 --- a/scripts/demo_app.py +++ b/scripts/demo_app.py @@ -50,6 +50,7 @@ class OtherSettings(pydantic.BaseSettings): other: str = 'other' +@fastapi_plugins.registered_configuration class AppSettings( OtherSettings, fastapi_plugins.ControlSettings, @@ -58,19 +59,29 @@ class AppSettings( MemcachedSettings, ): api_name: str = str(__name__) - # redis_type = fastapi_plugins.RedisType.sentinel - # redis_sentinels = 'localhost:26379' + + +@fastapi_plugins.registered_configuration(name='sentinel') +class AppSettingsSentinel(AppSettings): + redis_type = fastapi_plugins.RedisType.sentinel + redis_sentinels = 'localhost:26379' + + +@fastapi_plugins.registered_configuration_local +class AppSettingsLocal(AppSettings): + pass app = fastapi.FastAPI() -config = AppSettings() +config = fastapi_plugins.get_config() @app.get("/") async def root_get( cache: aioredis.Redis=fastapi.Depends(fastapi_plugins.depends_redis), + conf: pydantic.BaseSettings=fastapi.Depends(fastapi_plugins.depends_config) # noqa E501 ) -> typing.Dict: - return dict(ping=await cache.ping()) + return dict(ping=await cache.ping(), api_name=conf.api_name) @app.post("/jobs/schedule/") @@ -124,6 +135,8 @@ async def memcached_demo_post( @app.on_event('startup') async def on_startup() -> None: + await fastapi_plugins.config_plugin.init_app(app, config) + await fastapi_plugins.config_plugin.init() await memcached_plugin.init_app(app, config) await memcached_plugin.init() await fastapi_plugins.redis_plugin.init_app(app, config=config) @@ -145,3 +158,4 @@ async def on_shutdown() -> None: await fastapi_plugins.scheduler_plugin.terminate() await fastapi_plugins.redis_plugin.terminate() await memcached_plugin.terminate() + await fastapi_plugins.config_plugin.terminate() diff --git a/tests/conftest.py b/tests/conftest.py index 12e6855..02ccec2 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -37,3 +37,6 @@ def pytest_configure(config): config.addinivalue_line( "markers", "sentinel: mark test related to Redis Sentinel" ) + config.addinivalue_line( + "markers", "settings: mark test related to Settings and Configuration" + ) diff --git a/tests/test_settings.py b/tests/test_settings.py new file mode 100644 index 0000000..c631333 --- /dev/null +++ b/tests/test_settings.py @@ -0,0 +1,420 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# tests.test_settings +''' +:author: madkote +:contact: madkote(at)bluewin.ch +:copyright: Copyright 2021, madkote RES + +tests.test_settings +--------- +Module +''' + +from __future__ import absolute_import + +import asyncio +import os +import unittest + +import fastapi +# import pydantic +import pytest + +import fastapi_plugins + +from fastapi_plugins.settings import ConfigManager + +from . import VERSION +from . import d2json + +__all__ = [] +__author__ = 'madkote ' +__version__ = '.'.join(str(x) for x in VERSION) +__copyright__ = 'Copyright 2021, madkote RES' + + +@pytest.mark.settings +class TestSettings(unittest.TestCase): + def setUp(self): + fastapi_plugins.reset_config() + fastapi_plugins.get_config.cache_clear() + + def tearDown(self): + fastapi_plugins.reset_config() + fastapi_plugins.get_config.cache_clear() + + def test_manager_register(self): + class MyConfig(fastapi_plugins.PluginSettings): + api_name: str = 'API name' + + name = 'myconfig' + + m = ConfigManager() + m.register(name, MyConfig) + + exp = {name: MyConfig} + res = m._settings_map + self.assertTrue(res == exp, 'register failed') + + def test_manager_reset(self): + class MyConfig(fastapi_plugins.PluginSettings): + api_name: str = 'API name' + + name = 'myconfig' + + m = ConfigManager() + m.register(name, MyConfig) + + exp = {name: MyConfig} + res = m._settings_map + self.assertTrue(res == exp, 'register failed') + + m.reset() + exp = {} + res = m._settings_map + self.assertTrue(res == exp, 'reset failed') + + def test_manager_get_config(self): + class MyConfig(fastapi_plugins.PluginSettings): + api_name: str = 'API name' + + name = 'myconfig' + + m = ConfigManager() + m.register(name, MyConfig) + + exp = {name: MyConfig} + res = m._settings_map + self.assertTrue(res == exp, 'register failed') + + exp = d2json(MyConfig().dict()) + res = d2json(m.get_config(name).dict()) + self.assertTrue(res == exp, 'get configuration failed') + + def test_manager_get_config_default(self): + class MyConfig(fastapi_plugins.PluginSettings): + api_name: str = 'API name' + + name = fastapi_plugins.CONFIG_NAME_DEFAULT + + m = ConfigManager() + m.register(name, MyConfig) + + exp = {name: MyConfig} + res = m._settings_map + self.assertTrue(res == exp, 'register failed') + + exp = d2json(MyConfig().dict()) + res = d2json(m.get_config().dict()) + self.assertTrue(res == exp, 'get configuration failed') + + def test_manager_get_config_not_existing(self): + class MyConfig(fastapi_plugins.PluginSettings): + api_name: str = 'API name' + + name = 'myconfig' + + m = ConfigManager() + m.register(name, MyConfig) + + exp = {name: MyConfig} + res = m._settings_map + self.assertTrue(res == exp, 'register failed') + + try: + m.get_config() + except fastapi_plugins.ConfigError: + pass + else: + self.fail('configuration should not exist') + + def test_wrap_register_config(self): + try: + class MyConfig(fastapi_plugins.PluginSettings): + api_name: str = 'API name' + + fastapi_plugins.register_config(MyConfig) + exp = d2json(MyConfig().dict()) + res = d2json(fastapi_plugins.get_config().dict()) + self.assertTrue(res == exp, 'get configuration failed') + finally: + fastapi_plugins.reset_config() + + def test_wrap_register_config_docker(self): + try: + class MyConfig(fastapi_plugins.PluginSettings): + api_name: str = 'API name' + + # docker is default + fastapi_plugins.register_config_docker(MyConfig) + exp = d2json(MyConfig().dict()) + res = d2json(fastapi_plugins.get_config().dict()) + self.assertTrue(res == exp, 'get configuration failed') + # + exp = d2json(MyConfig().dict()) + res = d2json(fastapi_plugins.get_config(fastapi_plugins.CONFIG_NAME_DOCKER).dict()) # noqa E501 + self.assertTrue(res == exp, 'get configuration failed') + finally: + fastapi_plugins.reset_config() + + def test_wrap_register_config_local(self): + try: + class MyConfig(fastapi_plugins.PluginSettings): + api_name: str = 'API name' + + # + fastapi_plugins.register_config_local(MyConfig) + exp = d2json(MyConfig().dict()) + res = d2json(fastapi_plugins.get_config(fastapi_plugins.CONFIG_NAME_LOCAL).dict()) # noqa E501 + self.assertTrue(res == exp, 'get configuration failed') + # + os.environ[fastapi_plugins.DEFAULT_CONFIG_ENVVAR] = fastapi_plugins.CONFIG_NAME_LOCAL # noqa E501 + try: + exp = d2json(MyConfig().dict()) + res = d2json(fastapi_plugins.get_config().dict()) + self.assertTrue(res == exp, 'get configuration failed') + finally: + os.environ.pop(fastapi_plugins.DEFAULT_CONFIG_ENVVAR) + finally: + fastapi_plugins.reset_config() + + def test_wrap_register_config_test(self): + try: + class MyConfig(fastapi_plugins.PluginSettings): + api_name: str = 'API name' + + # + fastapi_plugins.register_config_test(MyConfig) + exp = d2json(MyConfig().dict()) + res = d2json(fastapi_plugins.get_config(fastapi_plugins.CONFIG_NAME_TEST).dict()) # noqa E501 + self.assertTrue(res == exp, 'get configuration failed') + # + os.environ[fastapi_plugins.DEFAULT_CONFIG_ENVVAR] = fastapi_plugins.CONFIG_NAME_TEST # noqa E501 + try: + exp = d2json(MyConfig().dict()) + res = d2json(fastapi_plugins.get_config().dict()) + self.assertTrue(res == exp, 'get configuration failed') + finally: + os.environ.pop(fastapi_plugins.DEFAULT_CONFIG_ENVVAR) + finally: + fastapi_plugins.reset_config() + + def test_wrap_register_config_by_name(self): + try: + class MyConfig(fastapi_plugins.PluginSettings): + api_name: str = 'API name' + + name = 'myconfig' + fastapi_plugins.register_config(MyConfig, name=name) + exp = d2json(MyConfig().dict()) + res = d2json(fastapi_plugins.get_config(name).dict()) + self.assertTrue(res == exp, 'get configuration failed') + finally: + fastapi_plugins.reset_config() + + def test_wrap_get_config(self): + try: + class MyConfig(fastapi_plugins.PluginSettings): + api_name: str = 'API name' + + fastapi_plugins.register_config(MyConfig) + exp = d2json(MyConfig().dict()) + res = d2json(fastapi_plugins.get_config().dict()) + self.assertTrue(res == exp, 'get configuration failed') + finally: + fastapi_plugins.reset_config() + + def test_wrap_get_config_by_name(self): + try: + class MyConfig(fastapi_plugins.PluginSettings): + api_name: str = 'API name' + + name = 'myconfig' + fastapi_plugins.register_config(MyConfig, name=name) + exp = d2json(MyConfig().dict()) + res = d2json(fastapi_plugins.get_config(name).dict()) + self.assertTrue(res == exp, 'get configuration failed') + finally: + fastapi_plugins.reset_config() + + def test_wrap_reset_config(self): + try: + class MyConfig(fastapi_plugins.PluginSettings): + api_name: str = 'API name' + + from fastapi_plugins.settings import _manager + + exp = {} + res = _manager._settings_map + self.assertTrue(res == exp, 'reset init failed') + + fastapi_plugins.register_config(MyConfig) + + exp = {fastapi_plugins.CONFIG_NAME_DOCKER: MyConfig} + res = _manager._settings_map + self.assertTrue(res == exp, 'reset register failed') + + fastapi_plugins.reset_config() + + exp = {} + res = _manager._settings_map + self.assertTrue(res == exp, 'reset failed') + finally: + fastapi_plugins.reset_config() + + def test_decorator_register_config(self): + try: + @fastapi_plugins.registered_configuration + class MyConfig(fastapi_plugins.PluginSettings): + api_name: str = 'API name' + + exp = d2json(MyConfig().dict()) + res = d2json(fastapi_plugins.get_config().dict()) + self.assertTrue(res == exp, 'get configuration failed: %s != %s' % (exp, res)) # noqa E501 + finally: + fastapi_plugins.reset_config() + + def test_decorator_register_config_docker(self): + try: + @fastapi_plugins.registered_configuration_docker + class MyConfig(fastapi_plugins.PluginSettings): + api_name: str = 'API name' + + # docker is default + exp = d2json(MyConfig().dict()) + res = d2json(fastapi_plugins.get_config().dict()) + self.assertTrue(res == exp, 'get configuration failed') + # + exp = d2json(MyConfig().dict()) + res = d2json(fastapi_plugins.get_config(fastapi_plugins.CONFIG_NAME_DOCKER).dict()) # noqa E501 + self.assertTrue(res == exp, 'get configuration failed') + finally: + fastapi_plugins.reset_config() + + def test_decorator_register_config_local(self): + try: + @fastapi_plugins.registered_configuration_local + class MyConfig(fastapi_plugins.PluginSettings): + api_name: str = 'API name' + + # + exp = d2json(MyConfig().dict()) + res = d2json(fastapi_plugins.get_config(fastapi_plugins.CONFIG_NAME_LOCAL).dict()) # noqa E501 + self.assertTrue(res == exp, 'get configuration failed') + # + os.environ[fastapi_plugins.DEFAULT_CONFIG_ENVVAR] = fastapi_plugins.CONFIG_NAME_LOCAL # noqa E501 + try: + exp = d2json(MyConfig().dict()) + res = d2json(fastapi_plugins.get_config().dict()) + self.assertTrue(res == exp, 'get configuration failed') + finally: + os.environ.pop(fastapi_plugins.DEFAULT_CONFIG_ENVVAR) + finally: + fastapi_plugins.reset_config() + + def test_decorator_register_config_test(self): + try: + @fastapi_plugins.registered_configuration_test + class MyConfig(fastapi_plugins.PluginSettings): + api_name: str = 'API name' + + # + exp = d2json(MyConfig().dict()) + res = d2json(fastapi_plugins.get_config(fastapi_plugins.CONFIG_NAME_TEST).dict()) # noqa E501 + self.assertTrue(res == exp, 'get configuration failed') + # + os.environ[fastapi_plugins.DEFAULT_CONFIG_ENVVAR] = fastapi_plugins.CONFIG_NAME_TEST # noqa E501 + try: + exp = d2json(MyConfig().dict()) + res = d2json(fastapi_plugins.get_config().dict()) + self.assertTrue(res == exp, 'get configuration failed') + finally: + os.environ.pop(fastapi_plugins.DEFAULT_CONFIG_ENVVAR) + finally: + fastapi_plugins.reset_config() + + def test_decorator_register_config_by_name(self): + try: + name = 'myconfig' + + @fastapi_plugins.registered_configuration(name=name) + class MyConfig(fastapi_plugins.PluginSettings): + api_name: str = 'API name' + + exp = d2json(MyConfig().dict()) + res = d2json(fastapi_plugins.get_config(name).dict()) + self.assertTrue(res == exp, 'get configuration failed') + finally: + fastapi_plugins.reset_config() + + def test_app_config(self): + async def _test(): + @fastapi_plugins.registered_configuration + class MyConfigDocker(fastapi_plugins.PluginSettings): + api_name: str = 'docker' + + @fastapi_plugins.registered_configuration_local + class MyConfigLocal(fastapi_plugins.PluginSettings): + api_name: str = 'local' + + app = fastapi.FastAPI() + config = fastapi_plugins.get_config() + + await fastapi_plugins.config_plugin.init_app(app=app, config=config) # noqa E501 + await fastapi_plugins.config_plugin.init() + + try: + c = await fastapi_plugins.config_plugin() + exp = d2json(MyConfigDocker().dict()) + res = d2json(c.dict()) + self.assertTrue(res == exp, 'get configuration failed') + finally: + await fastapi_plugins.config_plugin.terminate() + fastapi_plugins.reset_config() + + event_loop = asyncio.new_event_loop() + asyncio.set_event_loop(event_loop) + coro = asyncio.coroutine(_test) + event_loop.run_until_complete(coro()) + event_loop.close() + + def test_app_config_environ(self): + async def _test(): + os.environ[fastapi_plugins.DEFAULT_CONFIG_ENVVAR] = fastapi_plugins.CONFIG_NAME_LOCAL # noqa E501 + try: + @fastapi_plugins.registered_configuration + class MyConfigDocker(fastapi_plugins.PluginSettings): + api_name: str = 'docker' + + @fastapi_plugins.registered_configuration_local + class MyConfigLocal(fastapi_plugins.PluginSettings): + api_name: str = 'local' + + app = fastapi.FastAPI() + config = fastapi_plugins.get_config() + + await fastapi_plugins.config_plugin.init_app(app=app, config=config) # noqa E501 + await fastapi_plugins.config_plugin.init() + + try: + c = await fastapi_plugins.config_plugin() + exp = d2json(MyConfigLocal().dict()) + res = d2json(c.dict()) + self.assertTrue(res == exp, 'get configuration failed: %s != %s' % (exp, res)) # noqa E501 + finally: + await fastapi_plugins.config_plugin.terminate() + finally: + os.environ.pop(fastapi_plugins.DEFAULT_CONFIG_ENVVAR) + fastapi_plugins.reset_config() + + event_loop = asyncio.new_event_loop() + asyncio.set_event_loop(event_loop) + coro = asyncio.coroutine(_test) + event_loop.run_until_complete(coro()) + event_loop.close() + + +if __name__ == "__main__": + # import sys;sys.argv = ['', 'Test.testName'] + unittest.main()