FastAPI day 4
1. Multi application management
When we develop, a large number of routes are often involved. If all routes are written in one file, it is not conducive to us to modify a route and its processing function, and it is not conducive to the later maintenance and upgrading of the whole project. Moreover, the number of code lines in a file is too large, which will make the development particularly inconvenient. Therefore, it is necessary to divide the route into files (modularization).
I believe everyone who has studied the flash framework before should feel it. Isn't this the blueprint in flash? Yes, the APIRouter in FastAPI is very similar to the blueprint. It is also used to write routes in separate files, and it also needs to be finally registered in the main file. Don't say much. Use the example on the official website to understand.
First, let's look at the directory structure of the project
- Main is the final main file, which is the place where FastAPI instantiation and route registration are performed
- Routes and internal are two routing group files respectively
For simplicity, as in the case of two routes in routes, we directly instantiate APIRouter in__ init__ File, and then import them separately.
items.py
from ..routers import router
from fastapi import HTTPException
data={1:{"name":'aaa'},2:{"name":'bbb'}}
@router.get("/items/",tags=['items'])
async def read_data():
return data
@router.get("/items/{item_id}",tags=["items"])
async def read_item(item_id:int):
if item_id not in data:
raise HTTPException(status_code=404,detail="Item not Found!")
return {"item_id":item_id,"name":data[item_id]["name"]}
@router.put("/items/{item_id}",tags=["custom"],responses={403:{"description":"Operation forbidden"}})
async def update_item(item_id:int):
if item_id!=3:
raise HTTPException(
status_code=403,
detail="You can only update the item:3"
)
return {"item_id":item_id,"name":"ccc"}
users.py
from ..routers import router
@router.get("/users/",tags=["users"])
async def read_users():
return [
{"username":"aaa"},
{"username":"bbb"}
]
@router.get("/users/{username}",tags=['users'])
async def read_user(username:str):
return {"username":username}
admin.py
from fastapi import APIRouter
router=APIRouter()
@router.post("/")
async def update_admin():
return {"message":"Admin change"}
main.py
from fastapi import FastAPI
from .routers import items,users
from .internal import admin
app=FastAPI()
app.include_router(users.router)
app.include_router(items.router)
app.include_router(
admin.router,
prefix="/admin",
tags=['admin'],
)
@app.get("/")
async def root():
return {"message":"hello"}
Here is a brief summary of the relevant knowledge
- Prefix stands for prefix. For example, prefix=/index, followed by router Get ("AAA") actually requests / index/aaa
- Tags are tags, which are the names in Swagger
- Don't forget to use the router after writing it include_router() registers the written
2. Database related
In addition to storing data in files, we actually store more data in the database, which is more conducive to the management of a large amount of data (addition, deletion, modification and query). Therefore, let's take a look at the operations related to the database.
The first is the database connection part
import sqlalchemy
from databases import Database
DATABASE_URL="sqlite:///./test.db"
database=Database(DATABASE_URL)
sqlalchemy_engine=sqlalchemy.create_engine(DATABASE_URL)
def get_database()->Database:
return database
Then define the data model
from datetime import datetime
from typing import Optional
import sqlalchemy
from pydantic import BaseModel, Field
class PostBase(BaseModel):
title: str
content: str
publication_date: datetime = Field(default_factory=datetime.now)
class PostPartialUpdate(BaseModel):
title: Optional[str] = None
content: Optional[str] = None
class PostCreate(PostBase):
pass
class PostDB(PostBase):
id: int
metadata = sqlalchemy.MetaData()
posts=sqlalchemy.Table(
"posts",
metadata,
sqlalchemy.Column("id",sqlalchemy.Integer,primary_key=True,autoincrement=True),
sqlalchemy.Column("publication_date",sqlalchemy.DateTime(),nullable=False),
sqlalchemy.Column("title",sqlalchemy.String(length=255),nullable=False),
sqlalchemy.Column("content",sqlalchemy.Text(),nullable=False),
)
Firstly, the Pydantic model PostBase of the database is defined, and then the submission table format of the database is defined
This is similar to defining fields when creating a database table, including type, primary key, self increment, whether it can be empty, etc. this part is more inclined to the database foundation. btw I'm going to learn Mysql again in a while.
With the database connection and definition, you can create an interface for adding, deleting, modifying and querying
from typing import List, Tuple
from databases import Database
from fastapi import Depends, FastAPI, HTTPException, Query, status
from database import get_database, sqlalchemy_engine
from models import (
metadata,
posts,
PostDB,
PostCreate,
PostPartialUpdate,
)
app = FastAPI()
@app.on_event("startup")
async def startup():
await get_database().connect()
metadata.create_all(sqlalchemy_engine)
@app.on_event("shutdown")
async def shutdown():
await get_database().disconnect()
async def pagination(
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=0),
) -> Tuple[int, int]:
capped_limit = min(100, limit)
return (skip, capped_limit)
async def get_post_or_404(
id: int, database: Database = Depends(get_database)
) -> PostDB:
select_query = posts.select().where(posts.c.id == id)
raw_post = await database.fetch_one(select_query)
if raw_post is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
return PostDB(**raw_post)
@app.get("/posts")
async def list_posts(
pagination: Tuple[int, int] = Depends(pagination),
database: Database = Depends(get_database),
) -> List[PostDB]:
skip, limit = pagination
select_query = posts.select().offset(skip).limit(limit)
rows = await database.fetch_all(select_query)
results = [PostDB(**row) for row in rows]
return results
@app.get("/posts/{id}", response_model=PostDB)
async def get_post(post: PostDB = Depends(get_post_or_404)) -> PostDB:
return post
@app.post("/posts", response_model=PostDB, status_code=status.HTTP_201_CREATED)
async def create_post(
post: PostCreate, database: Database = Depends(get_database)
) -> PostDB:
insert_query = posts.insert().values(post.dict())
post_id = await database.execute(insert_query)
post_db = await get_post_or_404(post_id, database)
return post_db
@app.patch("/posts/{id}", response_model=PostDB)
async def update_post(
post_update: PostPartialUpdate,
post: PostDB = Depends(get_post_or_404),
database: Database = Depends(get_database),
) -> PostDB:
update_query = (
posts.update()
.where(posts.c.id == post.id)
.values(post_update.dict(exclude_unset=True))
)
post_id = await database.execute(update_query)
post_db = await get_post_or_404(post_id, database)
return post_db
@app.delete("/posts/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_post(
post: PostDB = Depends(get_post_or_404), database: Database = Depends(get_database)
):
delete_query = posts.delete().where(posts.c.id == post.id)
await database.execute(delete_query)
At the beginning, we set up two events
When we need to introduce dependency functions, get the database connection or close the database connection. Then set the operation of adding, deleting, modifying and querying in the route
insert_query = posts.insert().values(post.dict()) post_id = await database.execute(insert_query) post_db = await get_post_or_404(post_id, database)
Insert statement is to set the insert request statement first, and then execute this statement.
Others are similar, so we can realize the basic database addition, deletion, modification and query operations.