Я думал о шаблоне фабрики для приложений 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
Поскольку вы также отметили это с помощью
sanic
, я отвечу на этом фоне. Sanic является асинхронным и поэтому полагается на цикл событий. Цикл событий является ресурсом и, следовательно, не должен использоваться совместно между тестами, а должен создаваться заново для каждого из них. Следовательно, соединение с базой данных и т. Д. Также необходимо создавать для каждого теста, и его нельзя повторно использовать, поскольку оно асинхронно и зависит от цикла событий. Даже без асинхронной природы было бы проще создавать подключения к базе данных для каждого теста, потому что они имеют состояние (например, временные таблицы).В итоге я получил
create_app()
, который создает все, что позволяет мне создавать произвольное количество независимых приложений в тестовом прогоне. (Честно говоря, существуют некоторые глобальные ресурсы, такие как зарегистрированные прослушиватели событий, но их легко разобрать с помощью фабрик py.test.) Для тестируемости я бы попытался избежать глобальных ресурсов, которые создаются при импорте модуля. Хотя в больших и успешных проектах я видел другое.Я знаю, это не совсем однозначный ответ …
В итоге я получил create_app (), который создает все, что позволяет мне создавать произвольное количество независимых приложений в тестовом прогоне. — Я тоже этим занимаюсь. Возможность создавать независимые приложения мне кажется определением фабричной функции. — person timmwagener; 30.04.2018
Но в контексте приложения WSGI. это связано с некоторой суматохой, такой как сбор маршрутов в фабричной функции (в противном случае обычно регистрируется с помощью глобального объекта приложения) и то же самое, например, с задачами для сельдерея. Мне просто интересно, не слишком ли я над этим задумываюсь, потому что, похоже, это никого не беспокоит …!? — person timmwagener; 30.04.2018
До сих пор я использовал два метода регистрации маршрутов: (1) функция, которая принимает объект приложения и регистрирует все маршруты, или (2) с помощью flask_restful, который отделяет маршруты от объекта приложения. Однако первый имеет тенденцию быть немного загроможденным для более крупных проектов. — person timmwagener; 02.05.2018