FastAPI from the ground up

Chris Withers

Jump Trading

Who am I?

History

Static HTML
CGI scripts
Zope
DRF + React

What are the technologies?

REST

Async

Dependency Injection

REST

Representational State Transfer

GET /entry/?contains=foo

GET /entry/123

POST /entry/

PUT /entry/123

DELETE /entry/123

Async

Co-operative multitasking

async def get_item(name):
    client = httpx.AsyncClient()
    response = await client.get('https://.../'+name)
    return response.json()

Why should you hate it?

Dependency Injection

Say what you want, not how to get it.

def render(session: Session, name: str):
    item = Session.query(Item).filter_by(name=name)
    return make_json(item)

Why should you love it?

What are the options?

Django + DRF
Flask
Pyramid
Hug
FastAPI

Django + DRF

class UserSerializer(serializers.HyperlinkedModelSerializer):
    class Meta:
        model = User
        fields = ['url', 'username', 'email', 'groups']

class UserViewSet(viewsets.ModelViewSet):
    queryset = User.objects.all().order_by('-date_joined')
    serializer_class = UserSerializer

super flexible
web UI for free
django ORM
django settings

Flask

from flask import Flask, session, redirect, \
    url_for, escape, request

app = Flask(__name__)

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        session['username'] = request.form['username']
        return redirect(url_for('index'))
    return '''
        <form method="post">
            <p><input type=text name=username>
            <p><input type=submit value=Login>
        </form>
    '''

horrible globals
wsgi sync
no "DRF"?

Pyramid

def hello_world(request):
    return Response('Hello World!')

if __name__ == '__main__':
    with Configurator() as config:
        config.add_route('hello', '/')
        config.add_view(hello_world, route_name='hello')
        app = config.make_wsgi_app()
    server = make_server('0.0.0.0', 6543, app)
    server.serve_forever()

super flexible
very abstracted, but "no DRF" library problem
wsgi / sync only
(small community?)

Hug

@hug.get(examples='name=Timothy&age=26')
@hug.local()
def happy_birthday(
        name: hug.types.text,
        age: hug.types.number,
        hug_timer=3
):
    """Says happy birthday to a user"""
    return {
        'message': f'Happy {age} Birthday {name}!',
        'took': float(hug_timer)
    }

dependency injection
maintainer went awol
weird cli stuff?

Why FastAPI?

clean, new
dependency injection driven
optionally synchronous
automatic OpenAPI front end
great community
future proof

future proof: (websocket, graphql, etc)

URL mapping

app = FastAPI()

@app.get("/")
def root(...):
    return {"greeting": "Hello World"}

app.include_router(events.router, prefix="/events"...)

router = APIRouter()

@router.post("/", ..., status_code=201)
def create_object(...):
    ...

From Requests: Path Variables

GET /events/1234?detail=full Authorization: Token ABCD1234

@router.get("/event/{id}")
def get_object(id: int):
    ...

from fastapi import Path
from pydantic import Required

@router.get("events/{id}")
def get_object(id: int = Path(Required)):
    ...

From Requests: Query Parameters

GET /search/?text=something Authorization: Token ABCD1234

@router.get("/search")
def search(text:str = None, offset:int = 0, limit:int = 100):
    ...

from fastapi import Query

@router.get("/search")
def search(
    text: str = Query(default=None, max_length=20),
    offset: int = Query(0),
    limit: int = Query(100),
):

From Requests: Headers

GET /search/?text=something Authorization: Token ABCD1234

from fastapi import Header

@router.get("/search")
def search(authorization=Header(default=None)):
    ...

From Requests: Cookies

GET /search/?text=something Authorization: Token ABCD1234 🍪_suggest_author: Chris

from fastapi import Cookie

@router.get("/search/")
def search(_suggest_author=Cookie(default=None)):
    ...

From Requests: Forms

<form action="/files/"
      enctype="multipart/form-data" method="post">
<input name="token" type="text">
<input name="file" type="file">
<input type="submit">
</form>

from fastapi import File, Form, UploadFile
from pydantic import Required

@router.post("/files/")
async def create_file(
    token: str = Form(Required),
    file: UploadFile = File(Required),
):
    ...

From Requests: JSON Body

POST /events/ Authorization: Token ABCD1234

{
    'date': '2019-06-02',
    'type': 'DONE',
    'text': 'some stuff got done'
}

Data Validation: Pydantic

https://pydantic-docs.helpmanual.io/

from datetime import date as DateType
from enum import Enum
from pydantic import BaseModel

class Types(Enum):
    done = 'DONE'
    cancelled = 'CANCELLED'

class Event(BaseModel):
    date: DateType
    type: Types
    text: str

Data Validation: Pydantic

class Event(BaseModel):
    date: DateType
    type: Types
    text: str

>>> Event(date='2019-01-01', type='DONE', text='stuff') <Event date=datetime.date(2019, 1, 1) type=<Types.done: 'DONE'> text='stuff'>

>>> Event(date='2019-01-', text='some stuff') Traceback (most recent call last): ... pydantic.error_wrappers.ValidationError: 3 validation errors date invalid date format (type=type_error.date) type field required (type=value_error.missing)

From Requests: JSON Body

POST /events/ Authorization: Token ABCD1234

{
    'date': '2019-06-02',
    'type': 'DONE',
    'text': 'some stuff got done'
}

@router.post("/events/")
def create_object(event: Event = Required):
    ...

To Responses: Body

@app.get("/")
def root(...):
    return {"greeting": "Hello World"}

@app.post("/events/", response_model=Event, status_code=201)
def create_object(event: Event = Required):
    ...
    return {
        'date': '2019-06-02',
        'type': 'DONE',
        'text': 'some stuff got done'
    }

To Responses: Headers

from starlette.responses import Response

@app.get("/headers/")
def get_headers(response: Response):
    response.headers["X-Cat-Dog"] = "alone in the world"
    return {"message": "Hello World"}

from starlette.responses import JSONResponse

@app.get("/headers/")
def get_headers():
    content = {"message": "Hello World"}
    headers = {"X-Cat-Dog": "alone in the world"}
    return JSONResponse(content=content, headers=headers)

To Responses: Cookies

from starlette.responses import Response

@app.post("/cookie-and-object/")
def create_cookie(response: Response):
    response.set_cookie(key="fakesession",
                        value="fake-cookie-session-value")
    return {"message": "Come to the dark side"}

from starlette.responses import JSONResponse

@app.post("/cookie/")
def create_cookie():
    response = JSONResponse(content=...)
    response.set_cookie(key="fakesession",
                        value="fake-cookie-session-value")
    return response

OpenAPI Front End

Sync vs Async

app = FastAPI()

@app.get("/sync")
def root():
    return {"greeting": "Hello World"}

@app.get("/async")
async def root():
    return {"message": "Hello World"}

Dependencies

from fastapi import Depends, Security

@router.get("/{id}", response_model=EventFull)
def get_object(
    id: int,
    session: Session = Depends(db_session),
    current_user: User = Security(get_current_user)
):
    ...

can be sync or async
cached once per request

An application: Diary!

DID travel to pycon UK DID speak to Evil Util Co: -- Have promised it will be fixed tomorrow -- CANCELLED go to gym

Databases

from enum import Enum
from sqlalchemy import Column, Integer, Text, Date
from sqlalchemy.dialects.postgresql import ENUM
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Types(Enum):
    done = 'DONE'
    cancelled = 'CANCELLED'

class Event(Base):
    __tablename__ = 'entry'
    id = Column(Integer(), primary_key=True)
    date = Column(Date, nullable=False)
    type = Column(ENUM(Types, name='types_enum'))
    text = Column(Text, nullable=False)

SQLAlchemy
Where does the session come from?
Where do the credentials for that session come from?

Configuration

from configurator import Config
from pydantic import BaseModel, DSN

# schema
class DatabaseConfig(BaseModel):
    url: DSN

class AppConfig(BaseModel):
    testing: bool
    db: DatabaseConfig

# defaults
config = Config({
    'testing': False,
})

Configuration

def load_config(path=None):
    if config.testing:
        return
    # file
    if path is None:
        path = Path(__file__).resolve().parent / 'app.yml'
    config.merge(
        Config.from_path(path)
    )
    # env
    config.merge(os.environ, {
        'DB_URL': 'db.url',
    })
    # validate
    AppConfig(**config.data)
    return config

Pathlib! :-)

Configuration

from fastapi import FastAPI
from sqlalchemy import create_engine

app = FastAPI()

@app.on_event("startup")
def configure():
    load_config()
    Session.configure(bind=create_engine(config.db.url))

Databases Again

@app.get("/")
def root(session: Session = Depends(db_session)):
    return {"count": session.query(Event).count()}

from sqlalchemy.orm import sessionmaker

Session = sessionmaker()

def db_session(request: Request):
    return request.state.db

Databases Again

from starlette.requests import Request
from starlette.concurrency import run_in_threadpool

def finish_session(session):
    session.rollback()
    session.close()

@app.middleware('http')
async def make_db_session(request: Request, call_next):
    request.state.db = session = Session()
    response = await call_next(request)
    await run_in_threadpool(finish_session, session)
    return response

CRUD: Schemas

class EventNonPrimaryKey(BaseModel):
    date: DateType
    type: Types
    text: str

class EventFull(EventNonPrimaryKey):
    id: int

class EventList(BaseModel):
    items: List[EventFull]
    count: int
    prev: str = None
    next: str = None

CRUD: Create

@router.post("/", response_model=EventFull, status_code=201)
def create_object(
    event: EventNonPrimaryKey = Required,
    session: Session = Depends(db_session),
):
    """
    Create new Event.
    """
    with session.transaction:
        obj = Event(**event.dict())
        session.add(obj)
        session.flush()
        return simplify(obj)

explain simplify and why the code isn't there

CRUD: Read

@router.get("/{id}", response_model=EventFull)
def get_object(
    id: int,
    session: Session = Depends(db_session),
):
    """
    Get Event by ID.
    """
    with session.transaction:
        try:
            obj = session.query(Event).filter_by(id=id).one()
        except NoResultFound:
            raise HTTPException(status_code=404)
        else:
            return simplify(obj)

CRUD: List

@router.get("/", response_model=EventList, name='events_list')
def list_object(
    text: str = None, offset: int = Query(0), limit: int = 100,
    session: Session = Depends(db_session),
    request: Request = Required,
):
    with session.transaction:
        items = session.query(Event).order_by(Event.date.desc(), 'id')
        if text:
            items = items.filter(Event.text.ilike('%'+text.strip()+'%'))
        count = items.count()
        items = [EventFull(**simplify(i))
                 for i in items.offset(offset).limit(limit)]
        if len(items) != count:
                ...
                prev = url_for(request, ...)
                next = url_for(request, ...)
        else:
            prev = next = None
        return EventList(count=count, items=items, prev=prev, next=next)

CRUD: Update

@router.put("/{id}", response_model=EventFull)
def update_object(
    id: int,
    event: EventNonPrimaryKey = Required,
    session: Session = Depends(db_session),
):
    with session.transaction:
        try:
            obj = session.query(Event).filter_by(id=id).one()
        except NoResultFound:
            raise HTTPException(status_code=404)
        else:
            for key, value in event.dict().items():
                setattr(obj, key, value)
            return simplify(obj)

CRUD: Delete

@router.delete("/{id}", response_model=EventFull)
def delete_object(
    id: int,
    session: Session = Depends(db_session),
):
    """
    Delete an Event.
    """
    with session.transaction:
        try:
            obj = session.query(Event).filter_by(id=id).one()
        except NoResultFound:
            raise HTTPException(status_code=404)
        else:
            session.delete(obj)
            return simplify(obj)

Authentication

from fastapi import Depends, Security
from fastapi.security import HTTPBasic, HTTPBasicCredentials

security = HTTPBasic()

def get_current_user(
        credentials: HTTPBasicCredentials = Depends(security)
):
    return credentials.username

@router.get("/{id}", response_model=EventFull)
def get_object(
    id: int,
    session: Session = Depends(db_session),
    current_user: User = Security(get_current_user)
):
    ...

Supports OAuth2, all the other goodness Obviously need to actually check passwords!

Testing: The fixtures

@pytest.fixture(scope='session')
def client():
    with config.push({
        'testing': True,
        'db': {'url': os.environ['TEST_DB_URL']}
    }):
        with TestClient(app) as client:
            yield client

Testing: The fixtures

@pytest.fixture(scope='session')
def db(client):
    engine = Session.kw['bind']
    conn = engine.connect()
    transaction = conn.begin()
    try:
        Base.metadata.create_all(bind=conn,
                                 checkfirst=False)
        yield conn
    finally:
        transaction.rollback()
        Session.configure(bind=engine)

Testing: The fixtures

@pytest.fixture()
def session(db):
    transaction = db.begin_nested()
    try:
        Session.configure(bind=db)
        yield Session()
    finally:
        transaction.rollback()

Testing: The test

def test_create_full_data(session, client):
    response = client.post('/events/', json={
        'date': '2019-06-02',
        'type': 'DONE',
        'text': 'some stuff got done'
    })
    actual = session.query(Event).one()
    compare(actual.date, expected=date(2019, 6, 2))
    compare(response.json(), expected={
        'id': actual.id,
        'date': '2019-06-02',
        'type': 'DONE',
        'text': 'some stuff got done'
    })
    compare(response.status_code, expected=201)

remember to explain testfixtures!

What's still left to do?

Abstract out model helpers
Abstract out auth library
Support context manager dependencies

Questions?

Getting the talk materials:
https://cjw296.github.io/pycon-uk-2019-fastapi/
https://github.com/cjw296/diary/tree/master/backend

Fast API Documentation:

https://fastapi.tiangolo.com/

Getting hold of me:

chris@python.org / @chriswithers13

Space	Forward
Right, Down, Page Down	Next slide
Left, Up, Page Up	Previous slide
G	Go to slide number
P	Open presenter console
H	Toggle this help