Python使用FastAPI开发CRUD接口完整案例

前言

学无止境，无止境学。大家好，我是张大鹏，之前在抖音有5万多粉丝，不过现在不拍视频，专心写公众号了。笔者目前是高级Python工程师，之前是全栈工程师，主要擅长Golang和Python开发，对于Java，Vue，React也有一些研究。工作之余，喜欢学习和分享，希望能够通过此公众号'张大鹏520'，将自己学到的东西分享给大家，和大家一起交流，一起成长，一起进步。

今天要给大家分享的是<<Python使用FastAPI开发CRUD接口完整案例>>，这是一个非常基础但是功能比较完备的案例,包含一套CRUD接口.

效果预览

新增待办事项:

删除待办事项:

修改待办事项:

根据ID查询待办事项:

查询所有待办事项:

完整代码

以下是项目的完整代码,懒得复制的同学,可以在文章底部查看代码文件的获取方式哈.

from fastapi import FastAPI, APIRouter, Path
from pydantic import BaseModel

# 创建应用和路由
app = FastAPI()
todo_router = APIRouter()


# 创建模型
class TodoItem(BaseModel):
    item: str
    status: bool

    class Config:
        schema_extra = {
            'example': {
                'item': '代办任务名称',
                'status': False
            }
        }


class Todo(BaseModel):
    id: int
    item: TodoItem


# 创建数据库
todo_list: list[Todo] = []


# 创建路由
@todo_router.post('/todo')
async def todo_add(todo: Todo) -> dict:
    todo_list.append(todo)
    return {
        'todo': todo
    }


@todo_router.get('/todo')
async def todo_get_list() -> dict:
    return {
        'data': todo_list
    }


@todo_router.get('/todo/{todo_id}')
async def todo_get_by_id(todo_id: int = Path(..., title='任务ID')) -> dict:
    # 查找并返回
    for tt in todo_list:
        if tt.id == todo_id:
            return {
                'data': tt
            }

    # 查不到,返回空
    return {
        'data': None
    }


@todo_router.put('/todo/{todo_id}')
async def todo_update(todo_id: int = Path(..., title='任务ID'), todo: Todo = None) -> dict:
    # 查找并替换
    for i in range(len(todo_list)):
        if todo_list[i].id == todo_id:
            todo_list[i] = todo
            return {
                'data': todo_list[i]
            }

    # 查找失败
    return {
        'data': None
    }


@todo_router.delete('/todo/{todo_id}')
async def todo_delete_by_id(todo_id: int = Path(..., title='任务ID')) -> dict:
    # 查找并删除
    for i in range(len(todo_list)):
        if todo_list[i].id == todo_id:
            return {
                'data': todo_list.pop(i)
            }

    # 删除失败
    return {
        'data': None
    }


# 挂载路由
app.include_router(todo_router)

# 启动应用
if __name__ == '__main__':
    import uvicorn

    uvicorn.run('main:app', host='127.0.0.1', port=8888, reload=True)


请求模型

# 创建模型
class TodoItem(BaseModel):
    item: str
    status: bool

    class Config:
        schema_extra = {
            'example': {
                'item': '代办任务名称',
                'status': False
            }
        }


class Todo(BaseModel):
    id: int
    item: TodoItem

新增待办事项接口

@todo_router.post('/todo')
async def todo_add(todo: Todo) -> dict:
    todo_list.append(todo)
    return {
        'todo': todo
    }


删除待办事项接口

@todo_router.delete('/todo/{todo_id}')
async def todo_delete_by_id(todo_id: int = Path(..., title='任务ID')) -> dict:
    # 查找并删除
    for i in range(len(todo_list)):
        if todo_list[i].id == todo_id:
            return {
                'data': todo_list.pop(i)
            }

    # 删除失败
    return {
        'data': None
    }

更新待办事项接口

@todo_router.put('/todo/{todo_id}')
async def todo_update(todo_id: int = Path(..., title='任务ID'), todo: Todo = None) -> dict:
    # 查找并替换
    for i in range(len(todo_list)):
        if todo_list[i].id == todo_id:
            todo_list[i] = todo
            return {
                'data': todo_list[i]
            }

    # 查找失败
    return {
        'data': None
    }


根据ID查询待办事项接口

@todo_router.get('/todo/{todo_id}')
async def todo_get_by_id(todo_id: int = Path(..., title='任务ID')) -> dict:
    # 查找并返回
    for tt in todo_list:
        if tt.id == todo_id:
            return {
                'data': tt
            }

    # 查不到,返回空
    return {
        'data': None
    }

查询所有待办事项接口

@todo_router.get('/todo')
async def todo_get_list() -> dict:
    return {
        'data': todo_list
    }


结语