数据库
首页 > 数据库> > FastAPI第四天---数据库与模块化管理

FastAPI第四天---数据库与模块化管理

作者：互联网

文章目录

FastAPI第四天

1. 多应用程序管理

当我们开发的时候，往往会涉及到大量的路由，如果将所有的路由都写在一个文件中，不利于我们对于某个路由以及其处理函数进行修改，更不利于整个项目后期的维护升级。而且一个文件中代码行数过大还会使得开发尤为不便，因此需要将路由进行分文件（模块化）处理。

相信讲到这里，之前有学习过flask框架的应该都有感觉，这不就是flask中的蓝图吗？没错，FastAPI中的APIRouter与蓝图非常相似，同样都是为了分文件编写路由，也都是需要最终到主文件中进行注册。话不多说，借助官网的例子来理解一下。

首先来看看项目的目录结构

image-20220224161040441

为了简便化，像在routers中有两个路由的情况，我们直接把APIRouter实例化在__init__文件中，然后分别引入就好。

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"}

这里稍微总结一下有关的知识

  1. prefix代表前缀，比如prefix=/index，那么后面如果是router.get("aaa")请求的实际上是/index/aaa
  2. tags是标签，也就是Swagger中的名字

image-20220224162612792

  1. 编写好router后不要忘了使用.include_router()将写好的进行注册

2. 数据库相关

我们除了将数据存储到文件，其实更多的就是将数据存储到数据库中，这样更利于对较大量数据进行管理（增删改查），所以下面就来看看与数据库相关的操作。

首先是数据库连接部分

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

然后再来定义数据模型

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

首先定义了数据库的Pydantic模型PostBase，然后定义数据库的提交表格式image-20220224170939075

这里就类似于创建数据库表的时候定义字段一样，有类型、主键、自增、是否可以为空等等，这部分内容就更偏向于数据库基础，btw过段时间也准备重新学习一遍Mysql。

有了数据库连接以及定义之后，就可以创建增删改查的接口了

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)

image-20220224171333033

image-20220224171311620

image-20220224171344267

在开始，我们设置了两个事件

image-20220224171729313

在我们需要的时候引入依赖函数，得到数据库的连接或者关闭数据库连接。然后在路由中设置增删改查的操作

insert_query = posts.insert().values(post.dict())
post_id = await database.execute(insert_query)

post_db = await get_post_or_404(post_id, database)

插入语句就是先设置好插入请求语句，然后执行这条语句。

其他的也是类似，这样我们就能实现基本的数据库增删改查操作。

标签：database,FastAPI,get,posts,---,模块化,import,post,id
来源： https://blog.csdn.net/shelgi/article/details/123116678