Full Example¶
Let’s combine all of previous examples into a complete app.
FastAPI¶
import datetime
import typing as t
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from home.tables import Movie # An example Table
from piccolo.engine import engine_finder
from piccolo_admin.endpoints import create_admin
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.routing import Mount, Route
from piccolo_api.crud.serializers import create_pydantic_model
from piccolo_api.csrf.middleware import CSRFMiddleware
from piccolo_api.openapi.endpoints import swagger_ui
from piccolo_api.session_auth.endpoints import session_login, session_logout
from piccolo_api.session_auth.middleware import SessionsAuthBackend
# The main app with public endpoints, which will be served by Uvicorn
app = FastAPI(
routes=[
# If we want to use Piccolo admin:
Mount(
"/admin/",
create_admin(
tables=[Movie],
# Required when running under HTTPS:
# allowed_hosts=['my_site.com']
),
),
# Session Auth login:
Mount(
"/login/",
session_login(redirect_to="/private/docs/"),
),
],
docs_url=None,
redoc_url=None,
)
# The private app with SessionAuth protected endpoints
private_app = FastAPI(
routes=[
Route("/logout/", session_logout(redirect_to="/")),
# We use a custom Swagger docs endpoint instead of the default FastAPI
# one, because this one supports CSRF middleware:
Mount("/docs/", swagger_ui(schema_url="/private/openapi.json")),
],
middleware=[
Middleware(
AuthenticationMiddleware,
backend=SessionsAuthBackend(
increase_expiry=datetime.timedelta(minutes=30)
),
),
# CSRF middleware provides additional protection for older browsers, as
# we're using cookies.
Middleware(CSRFMiddleware, allow_form_param=True),
],
# We disable the default Swagger docs, as we have a custom one (see above).
docs_url=None,
redoc_url=None,
)
# Mount our private app within the main app.
app.mount("/private/", private_app)
###############################################################################
# Example FastAPI endpoints and Pydantic models.
MovieModelIn: t.Any = create_pydantic_model(
table=Movie, model_name="MovieModelIn"
)
MovieModelOut: t.Any = create_pydantic_model(
table=Movie, include_default_columns=True, model_name="MovieModelOut"
)
@private_app.get("/movies/", response_model=t.List[MovieModelOut])
async def movies():
return await Movie.select().order_by(Movie._meta.primary_key)
@private_app.post("/movies/", response_model=MovieModelOut)
async def create_movie(movie_model: MovieModelIn):
movie = Movie(**movie_model.dict())
await movie.save()
return MovieModelOut(**movie.to_dict())
@private_app.put("/movies/{movie_id}/", response_model=MovieModelOut)
async def update_movie(movie_id: int, movie_model: MovieModelIn):
movie = await Movie.objects().get(Movie._meta.primary_key == movie_id)
if not movie:
return JSONResponse({}, status_code=404)
for key, value in movie_model.dict().items():
setattr(movie, key, value)
await movie.save().run()
return MovieModelOut(**movie.to_dict())
@private_app.delete("/movies/{movie_id}/")
async def delete_movie(movie_id: int):
movie = (
await Movie.objects()
.where(Movie._meta.primary_key == movie_id)
.first()
)
if not movie:
return JSONResponse({}, status_code=404)
await movie.remove()
return JSONResponse({})
###############################################################################
# This is optional - it's for creating a connection pool.
@app.on_event("startup")
async def open_database_connection_pool():
try:
engine = engine_finder()
await engine.start_connection_pool()
except Exception:
print("Unable to connect to the database")
@app.on_event("shutdown")
async def close_database_connection_pool():
try:
engine = engine_finder()
await engine.close_connection_pool()
except Exception:
print("Unable to connect to the database")
###############################################################################
if __name__ == "__main__":
import uvicorn
uvicorn.run(app)
Starlette¶
Is almost identical to the FastAPI example - just replace FastAPI
with
Starlette
, and use Starlette’s HTTPEndpoint
for your endpoints.