Create a CRUD API using FastAPI
FastAPI is an asynchronous framework for creating python WEB APIs, which means that you can easily create high-performance APIs.
Try it by referring to Tutorial.
1. File structure of the application to be created
Create a CRUD for the users table. I create about 6 files, but each has a small amount of code.
fastapi-crud-example
│ db.py
│ main.py
│
├─users
│ │ models.py
│ │ router.py
│ └─ schemas.py
│
└─utils
dbutils.py
- source
(1) db.py
sqlalchemy is used to define models and generate queries. Use Databases to access the database.
db.py
import databases
import sqlalchemy
DATABASE = 'postgresql'
USER = 'testuser'
PASSWORD = 'secret'
HOST = 'localhost'
PORT = '5432'
DB_NAME = 'testdb'
DATABASE_URL = '{}://{}:{}@{}:{}/{}'.format(DATABASE, USER, PASSWORD, HOST, PORT, DB_NAME)
# databases
database = databases.Database(DATABASE_URL, min_size=5, max_size=20)
ECHO_LOG = False
engine = sqlalchemy.create_engine(DATABASE_URL, echo=ECHO_LOG)
metadata = sqlalchemy.MetaData()
(2) users/models.py
models.py
import sqlalchemy
from db import metadata, engine
users = sqlalchemy.Table(
"users",
metadata,
sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True, index=True),
sqlalchemy.Column("username", sqlalchemy.String, index=True),
sqlalchemy.Column("email", sqlalchemy.String, index=True),
sqlalchemy.Column("hashed_password", sqlalchemy.String),
sqlalchemy.Column("is_active", sqlalchemy.Boolean(), default=True),
sqlalchemy.Column("is_superuser", sqlalchemy.Boolean(), default=False)
)
metadata.create_all(bind=engine)
(3) users/schemas.py
Use the model in pydantic to define the model for crud as well.
schemas.py
from pydantic import BaseModel
#request model for insert. id(automatic numbering)Is not defined because no input is required.
class UserCreate(BaseModel):
username: str
email: str
password: str
is_active: bool
is_superuser: bool
#request model for update
class UserUpdate(BaseModel):
id : int
username: str
email: str
password: str
is_active: bool
is_superuser: bool
#request model for select. In select, no password is required, so it is not defined.
class UserSelect(BaseModel):
username: str
email: str
is_active: bool
is_superuser: bool
(4) users/router.py
It will be the main part of crud. All of select, insert, update and delete are implemented in less than 10 lines. It is very nice that the amount of coding is small.
router.py
import hashlib
from fastapi import APIRouter, Depends
from typing import List
from starlette.requests import Request
from .models import users
from .schemas import UserCreate, UserUpdate, UserSelect
from databases import Database
from utils.dbutils import get_connection
router = APIRouter()
#The entered password (plain text) is hashed and returned.
def get_users_insert_dict(user):
pwhash=hashlib.sha256(user.password.encode('utf-8')).hexdigest()
values=user.dict()
values.pop("password")
values["hashed_password"]=pwhash
return values
#Search all users and return the list of "User Select" as json.
@router.get("/users/", response_model=List[UserSelect])
async def users_findall(request: Request, database: Database = Depends(get_connection)):
query = users.select()
return await database.fetch_all(query)
#Search users by id and return "User Select" as json.
@router.get("/users/find", response_model=UserSelect)
async def users_findone(id: int, database: Database = Depends(get_connection)):
query = users.select().where(users.columns.id==id)
return await database.fetch_one(query)
#Register new users.
@router.post("/users/create", response_model=UserSelect)
async def users_create(user: UserCreate, database: Database = Depends(get_connection)):
#validator omitted
query = users.insert()
values = get_users_insert_dict(user)
ret = await database.execute(query, values)
return {**user.dict()}
#Update users.
@router.post("/users/update", response_model=UserSelect)
async def users_update(user: UserUpdate, database: Database = Depends(get_connection)):
#validator omitted
query = users.update().where(users.columns.id==user.id)
values=get_users_insert_dict(user)
ret = await database.execute(query, values)
return {**user.dict()}
#Delete users.
@router.post("/users/delete")
async def users_delete(user: UserUpdate, database: Database = Depends(get_connection)):
query = users.delete().where(users.columns.id==user.id)
ret = await database.execute(query)
return {"result": "delete success"}
(5) utils/dbutils.py
dbutils.py
from starlette.requests import Request
#Connection stored in request by middleware(Database object)Returns.
def get_connection(request: Request):
return request.state.connection
(6) main.py
main.py
from fastapi import FastAPI
from db import database
from users.router import router as userrouter
from starlette.requests import Request
app = FastAPI()
#Connect to Database at startup.
@app.on_event("startup")
async def startup():
await database.connect()
#Disconnect Database when finished.
@app.on_event("shutdown")
async def shutdown():
await database.disconnect()
#Register the users router.
app.include_router(userrouter)
# middleware state.Set the database object in connection.
@app.middleware("http")
async def db_session_middleware(request: Request, call_next):
request.state.connection = database
response = await call_next(request)
return response
3. Start and confirm
Enter "uvicorn main: app --reload" from powershell and enter
uvicorn main:app --reload
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [14108]
INFO: Started server process [21068]
INFO: Waiting for application startup.
INFO: Connected to database postgresql://testuser:********@localhost:5432/fastapidb
4. Check with Swagger
You can test the api with the Swagger UI. It's convenient.
Try to access http://127.0.0.1:8000/docs with chrome.
5. Finally
I got the impression that FastAPI is a freame work specialized for api, but you can also use a template engine using jinja2 etc., and it also has an authentication function such as oauth2. FastAPI It feels good.
Recommended Posts