Я думал о шаблоне фабрики для приложений 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
Источник: