Source code for molten.contrib.sqlalchemy
# This file is a part of molten.
#
# Copyright (C) 2018 CLEARTYPE SRL <[email protected]>
#
# molten is free software; you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or (at
# your option) any later version.
#
# molten is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from collections import namedtuple
from inspect import Parameter
from typing import Any, Callable, NewType, Optional
from molten import DependencyResolver, Settings
try:
from sqlalchemy import create_engine # type: ignore
from sqlalchemy.orm import Session, sessionmaker
except ImportError: # pragma: no cover
raise ImportError("'sqlalchemy' package missing. Run 'pip install sqlalchemy'.")
#: The type of session factories.
SessionFactory = NewType("SessionFactory", sessionmaker) # type: ignore
#: A named tuple containing an instantiated SQLAlchemy ``engine``
#: object and the ``session_factory``.
EngineData = namedtuple("EngineData", "engine, session_factory")
[docs]class SQLAlchemyEngineComponent:
"""A component that sets up an SQLAlchemy Engine. This component
depends on the availability of a :class:`molten.Settings`
component.
Your settings dictionary must contain a ``database_engine_dsn``
setting pointing at the database to use. Additionally, you may
provide a ``database_engine_params`` setting representing
dictionary data that will be passed directly to
``sqlalchemy.create_engine``.
Examples:
>>> from molten import App
>>> from molten.contrib.sqlalchemy import SQLAlchemyEngineComponent, SQLAlchemySessionComponent, SQLAlchemyMiddleware
>>> from molten.contrib.toml_settings import TOMLSettingsComponent
>>> app = App(
... components=[
... TOMLSettingsComponent(),
... SQLAlchemyEngineComponent(),
... SQLAlchemySessionComponent(),
... ],
... middleware=[SQLAlchemyMiddleware()],
... )
"""
is_cacheable = True
is_singleton = True
def can_handle_parameter(self, parameter: Parameter) -> bool:
return parameter.annotation is EngineData
def resolve(self, settings: Settings) -> EngineData:
engine = create_engine(
settings.strict_get("database_engine_dsn"),
**settings.get("database_engine_params", {}),
)
session_factory = sessionmaker()
session_factory.configure(bind=engine)
return EngineData(engine, session_factory)
[docs]class SQLAlchemySessionComponent:
"""A component that creates and injects SQLAlchemy sessions.
Examples:
>>> def find_todos(session: Session) -> List[Todo]:
... todos = session.query(TodoModel).all()
... ...
"""
is_cacheable = True
is_singleton = False
def can_handle_parameter(self, parameter: Parameter) -> bool:
return parameter.annotation is Session
def resolve(self, engine_data: EngineData) -> Session: # type: ignore
return engine_data.session_factory()
[docs]class SQLAlchemyMiddleware:
"""A middleware that automatically commits SQLAlchemy sessions on
handler success and automatically rolls back sessions on handler
failure.
Sessions are only instantiated and operated upon if the handler or
any other middleware has requested an SQLAlchemy session object
via DI. This means that handlers that don't request a Session
object don't automatically connect to the Database.
"""
def __call__(self, handler: Callable[..., Any]) -> Callable[..., Any]:
def middleware(resolver: DependencyResolver) -> Any:
session = None
try:
response = handler()
session = get_optional_session(resolver)
if session is not None:
session.commit()
return response
except Exception:
session = get_optional_session(resolver)
if session is not None:
session.rollback()
raise
finally:
if session is not None:
session.close()
return middleware
def get_optional_session(resolver: DependencyResolver) -> Optional[Session]: # type: ignore
"""Get a session object from the resolver iff one was previously
requested. Returns None if no function has requested a session so
far.
"""
for component, value in resolver.instances.items():
if type(component) is SQLAlchemySessionComponent:
return value
return None