Hello, good evening, good morning. This is Shimabukuro from K.S. Rogers.
I wrote an article like Previously the next hegemony of Python's web framework is this, but since then I've been an engineer away from Python for a while. By the way, a framework called FastAPI has appeared!
To be honest, I'm late because I don't know what number to brew, but let me introduce it.
According to the Official Top Page,
It says, but it's compatible because it's the ʻOpenAPI standard at the bottom (compatible with Swagger)` This is too great. FastAPI, in fact, can automatically generate Swagger (document)!
After installing fastapi
and ʻuvicorn` with pip, try touching them immediately.
By the way, standard pip does not have environment isolation, so it is recommended to use pipenv.
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}
After creating the above with main.py
, start it with ʻuvicorn main: app --reload. If you check
http://127.0.0.1:8000/docs` ...
Yes! Nice! !!
If you try setting path parameter
...
Path parameters
@app.get("/{name}")
async def root():
return {"message": "Hello " + name}
It will be reflected automatically. Documents are created automatically, which is the strongest. This is a function that is too convenient for us, who place particular importance on documents.
Query Parameters
@app.get("/")
async def read_item(name: str = ""):
return {"message": "Hello " + name}
Requests Use Model.
from fastapi import FastAPI
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str = None
app = FastAPI()
@app.post("/items/")
async def create_item(item: Item):
return item
Furthermore, validation can be specified for the property in the Model class, so the handling is very good.
This is a similar structure to the Django
Model.
class Item(BaseModel):
name: str = Field(None, min_length=2, max_length=5)
description: str = Field(None, max_length=100)
CRUD (Crate, Read, Update, Delete) Let's roughly write the basic CRUD. Since it is a partial excerpt, please refer to the writing link for details.
https://github.com/shuto-S/sample-fast-api
DB Connection & Migration
Use databases
and sqlalchemy
to connect to and migrate to the DB.
This time, we will use SQLite
, which is easy to handle, for the DB.
Write around the connection quickly,
import databases
import sqlalchemy
DATABASE_URL = "sqlite:///./db.sqlite3"
database = databases.Database(DATABASE_URL)
engine = sqlalchemy.create_engine(DATABASE_URL, echo=False)
metadata = sqlalchemy.MetaData()
Schemas Describe the table definition and execute the same file.
import sqlalchemy
from db import metadata, engine
items = sqlalchemy.Table(
"items",
metadata,
sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
sqlalchemy.Column("name", sqlalchemy.String),
sqlalchemy.Column("description", sqlalchemy.String),
)
metadata.create_all(engine)
Main
main.py
also repaired
from fastapi import FastAPI
from db import database
from starlette.requests import Request
from routers import items
app = FastAPI()
@app.on_event("startup")
async def startup():
#DB connection started
await database.connect()
@app.on_event("shutdown")
async def shutdown():
#DB connection disconnection
await database.disconnect()
#Register routers
app.include_router(items.router)
#Embed a DB connection with middleware (so that it can be obtained with a router)
@app.middleware("http")
async def db_session_middleware(request: Request, call_next):
request.state.connection = database
response = await call_next(request)
return response
Routers
router
is also divided into files so that it can be described for each endpoint.
from fastapi import APIRouter, Depends, HTTPException
from typing import List
from databases import Database
from starlette.status import HTTP_204_NO_CONTENT
from utils import get_db_connection
from schemas import items
from models.item import ItemModel
router = APIRouter()
@router.get("/items", tags=["items"], response_model=List[ItemModel])
async def list_item(database: Database = Depends(get_db_connection)):
query = items.select()
return await database.fetch_all(query)
@router.post("/items", tags=["items"], response_model=ItemModel)
async def create_item(data: ItemModel, database: Database = Depends(get_db_connection)):
query = items.insert()
await database.execute(query, data.dict())
return {**data.dict()}
@router.patch("/items/{item_id}", tags=["items"], response_model=ItemModel)
async def update_item(item_id: int, data: ItemModel, database: Database = Depends(get_db_connection)):
query = items.update().where(items.columns.id==item_id)
ret = await database.execute(query, data.dict())
if not ret:
raise HTTPException(status_code=404, detail="Not Found")
return {**data.dict()}
@router.delete("/items/{item_id}", tags=["items"], status_code=HTTP_204_NO_CONTENT)
async def delete_item(item_id: int, database: Database = Depends(get_db_connection)):
query = items.delete().where(items.columns.id==item_id)
ret = await database.execute(query)
if not ret:
raise HTTPException(status_code=404, detail="Not Found")
How is it? Isn't document auto-generation a pretty attractive feature? I think Responder is also a pretty good framework, but FastAPI feels less quirky and easier to write.
By the way, we also post company blogs in addition to our company and Tech, so if you are interested, please do. https://www.wantedly.com/companies/ks-rogers
Recommended Posts