SQLAlchemy integration for API Star
Branch: master
Clone or download
Fetching latest commit…
Cannot retrieve the latest commit at this time.
Permalink
Type Name Latest commit message Commit time
Failed to load latest commit information.
apistar_sqlalchemy
tests
.gitignore
.pre-commit-config.yaml
.prospector.yaml
.travis.yml
CHANGELOG.rst
CONTRIBUTORS.rst
LICENSE
MANIFEST.in
Pipfile
Pipfile.lock
README.rst
build.py
setup.cfg
setup.py
tox.ini

README.rst

API Star SQLAlchemy

build status coverage version

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)