一段时间以来，我一直在考虑Flask 文档推荐的WSGI 应用程序的工厂模式。特别是那些通常显示为使用在模块导入时创建的对象的函数，如示例中所示，而不是在工厂函数中创建的对象。db
理想情况下，工厂功能会重新创建_everything_，还是对db引擎等对象没有意义？
（我在这里考虑更清晰的分离和更好的可测试性。）
这是一些代码，我试图在其中完成为 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