FastAPI day 4 - database and modular management

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.

from ..routers import router
from fastapi import HTTPException


async def read_data():
    return data

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(
            detail="You can only update the item:3"

    return {"item_id":item_id,"name":"ccc"}

from ..routers import router

async def read_users():
    return [

async def read_user(username:str):
    return {"username":username}

from fastapi import APIRouter

async def update_admin():
    return {"message":"Admin change"}

from fastapi import FastAPI
from .routers import items,users
from .internal import admin



async def root():
    return {"message":"hello"}

Here is a brief summary of the relevant knowledge

  1. Prefix stands for prefix. For example, prefix=/index, followed by router Get ("AAA") actually requests / index/aaa
  2. Tags are tags, which are the names in Swagger

  1. 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


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(

class PostPartialUpdate(BaseModel):
    title: Optional[str] = None
    content: Optional[str] = None

class PostCreate(PostBase):

class PostDB(PostBase):
    id: int

metadata = sqlalchemy.MetaData()


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 (

app = FastAPI()

async def startup():
    await get_database().connect()

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 = == 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)

async def list_posts(
    pagination: Tuple[int, int] = Depends(pagination),
    database: Database = Depends(get_database),
) -> List[PostDB]:
    skip, limit = pagination
    select_query =
    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"/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 = (
        .where( ==
    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( ==
    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.

Keywords: Python Database FastAPI

Added by webstyler on Thu, 24 Feb 2022 11:32:44 +0200