Вопросы

Шаблон фабрики приложений WSGI и время импорта созданных объектов уровня модуля

Я думал о шаблоне фабрики для приложений WSGI, в соответствии с рекомендациями документации Flask. В частности, о тех функциях, которые обычно используются для использования объектов, которые были созданы во время импорта модуля, например db в примере, в отличие от тех, которые были созданы в фабричной функции.

Будет ли в идеале фабричная функция создавать _everything_ заново или это не имело бы смысла для таких объектов, как db engine? (здесь я думаю о более чистом разделении и лучшей тестируемости .)

Вот код, в котором я пытаюсь создать все необходимые объекты для приложения wsgi. в своей заводской функции.

# factories.py
def create_app(config, engine=None):
    """Create WSGI application to be called by WSGI server. Full factory function
    that takes care to deliver entirely new WSGI application instance with all
    new member objects like database engine etc.

    Args:
        config (dict): Dict to update the wsgi app. configuration.
        engine (SQLAlchemy engine): Database engine to use.
    """

    # flask app
    app = Flask(__name__)  # should be package name instead of __name__ acc. to docs
    app.config.update(config)

    # create blueprint
    blueprint = ViewRegistrationBlueprint('blueprint', __name__, )
    # request teardown behaviour, always called, even on unhandled exceptions

    # register views for blueprint
    from myapp.views import hello_world
    # dynamically scrapes module and registers methods as views
    blueprint.register_routes(hello_world)

    # create engine and request scoped session for current configuration and store
    # on wsgi app
    if (engine is not None):

        # delivers transactional scope when called
        RequestScopedSession = scoped_session(
            sessionmaker(bind=engine),
            scopefunc=flask_request_scope_func
        )

        def request_scoped_session_teardown(*args, **kwargs):
            """Function to register and call by the framework when a request is finished
            and the session should be removed.
            """
            # wrapped in try/finally to make sure no error collapses call stack here
            try:
                RequestScopedSession.remove()  # rollback all pending changes, close and return conn. to pool
            except Exception as exception_instance:
                msg = "Error removing session in request teardown.\n{}"
                msg = msg.format(exception_instance)
                logger.error(msg)
            finally:
                pass

        app.config["session"] = RequestScopedSession
        blueprint.teardown_request(request_scoped_session_teardown)

    # register blueprint
    app.register_blueprint(blueprint)

    return app


def create_engine(config):
    """Create database engine from configuration

    Args:
        config (dict): Dict used to assemble the connection string.
    """

    # connection_string
    connection_string = "{connector}://{user}:{password}@{host}/{schema}"
    connection_string = connection_string.format(**config)

    # database engine
    return sqlalchemy_create_engine(
        connection_string,
        pool_size=10,
        pool_recycle=7200,
        max_overflow=0,
        echo=True
    )
# wsgi.py (served by WSGI server)
from myapp.factories import create_app
from myapp.factories import create_engine
from myapp.configuration.config import Config

config = Config()

engine = create_engine(config.database_config)
app = create_app(config.application_config, engine=engine)
# conftest.py
from myapp.factories import create_app
from myapp.factories import create_engine
from myapp.configuration.config import Config

@pytest.fixture
def app():
    config = TestConfig()
    engine = create_engine(config.database_config)
    app = create_app(config.application_config, engine=engine)
    with app.app_context():
        yield app
Читать:
Дамп базы данных mysql в резервную копию открытого текста (CSV) из командной строки

Похожие записи

Переименование файла Git. История доступна в cmd-строке, но не в интерфейсе github?

admin

ошибка R10 при развертывании приложения Flask с докером на heroku

admin

DiscordAPIError: Unknown Message (Бот для назначения ролей в сообщении)

admin

Очистить область виджета ячейки в блокноте Jupyter из блокнота

admin

Vue — переходная группа не работает с разными типами компонентов

admin

Traefik как Kubernetes Ingress в Azure — назначение статического IP-адреса службе Traefik

admin