apistar-sqlalchemy 0.3.2

SQLAlchemy integration for API Star.

Version:

0.3.2

Status:

Production/Stable

Author:

José Antonio Perdiguero López

SQLAlchemy integration for API Star.

Features

This library provides components for injecting SQLAlchemy ORM sessions into your views and event_hooks to handle commit/rollback behavior based on exceptions in your views.

Quick start

Install API Star SQLAlchemy:

pip install apistar-sqlalchemy

Create an API Star application adding components and event hooks:

from apistar_sqlalchemy.components import SQLAlchemySessionComponent
from apistar_sqlalchemy.event_hooks import SQLAlchemyTransactionHook

routes = []

components = [
    SQLAlchemySessionComponent(url='sqlite://'),
]

event_hooks = [
    SQLAlchemyTransactionHook(),
]

app = App(routes=routes, components=components, event_hooks=event_hooks)

Now you can inject SQLAlchemy Session into your views:

from sqlalchemy.orm import Session

def sqlalchemy_view(session: Session):
    # do something
    return {'message': 'something done'}

Forget about commit and rollback because there is an event hook that will handle it for you.

Full example

from typing import List

from apistar import App, Route, http, types, validators
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import Session

from apistar_sqlalchemy.components import SQLAlchemySessionComponent
from apistar_sqlalchemy.event_hooks import SQLAlchemyTransactionHook
from apistar_sqlalchemy import database


class PuppyModel(database.Base):
    __tablename__ = "Puppy"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String)


class PuppyType(types.Type):
    id = validators.Integer(allow_null=True, default=None)
    name = validators.String()


def list_puppies(session: Session) -> List[PuppyType]:
    return [PuppyType(puppy) for puppy in session.query(PuppyModel).all()]


def create_puppy(session: Session, puppy: PuppyType) -> http.JSONResponse:
    model = PuppyModel(**puppy)
    session.add(model)
    session.flush()
    return http.JSONResponse(PuppyType(model), status_code=201)


routes = [
    Route('/puppy/', 'POST', create_puppy),
    Route('/puppy/', 'GET', list_puppies),
]

components = [
    SQLAlchemySessionComponent(url='sqlite://'),
]

event_hooks = [
    SQLAlchemyTransactionHook(),
]

app = App(routes=routes, components=components, event_hooks=event_hooks)