Apistar 0.4 Extended JSON Response, SQLAlchemy Component, and SQLAlchemy Hook

apistar_util.py

import datetime as dt
import decimal
import typing
from apistar import types
from apistar.http import JSONResponse, Response
from apistar.server.components import Component
from sqlalchemy.engine import Engine
from sqlalchemy.orm import sessionmaker, Session, scoped_session


class ExtJSONResponse(JSONResponse):
    """JSON Response with support for ISO 8601 datetime serialization and Decimal to float casting"""

    def default(self, obj: typing.Any) -> typing.Any:
        if isinstance(obj, types.Type):
            return dict(obj)
        if isinstance(obj, dt.datetime):
            return obj.isoformat()
        elif isinstance(obj, decimal.Decimal):
            return float(obj)
        error = "Object of type '%s' is not JSON serializable."
        return TypeError(error % type(obj).__name_)


DBSession = scoped_session(sessionmaker())


class SQLAlchemySession(Component):
    def __init__(self, engine=None):
        if not isinstance(engine, Engine):
            raise ValueError('SQLAlchemySession must be instantiated with a sqlalchemy.engine.Engine object')
        self.engine = engine
        DBSession.configure(bind=self.engine)

    def resolve(self) -> Session:
        return DBSession()


class SQLAlchemyHook:
    def on_request(self, session: Session):
        print('Hitting SQLAlchemy on_request Event Hook')
        print('Session ID:{}'.format(id(session)))
        return

    def on_response(self, session: Session, response: Response):
        print('Hitting SQLAlchemy on_response Event Hook')
        print('Session ID:{}'.format(id(session)))
        DBSession.remove()
        return response

    def on_error(self, session: Session, response: Response):
        print('Session ID:{}'.format(id(session)))
        session.rollback()
        DBSession.remove()
        return response