本案例使用 FastAPI 框架构建了一个简单的 员工管理系统，主要提供 添加、查询、修改、删除 员工信息的基本功能。数据存储使用 SQLite，并借助 SQLModel 进行 ORM 操作。整个系统采用 RESTful API 设计，前端或其他客户端可通过 HTTP 请求与后端交互。此外，案例中包含了一个 requests 客户端，用于演示 API 的调用方式。

conda create -n fastapi-env python=3.10
pip install uvicorn fastapi
pip install sqlmodel

1. 接口开发

系统共设计了四个核心接口：

• 添加员工信息 (POST /add)：接受员工信息，添加到数据库中并返回新增的员工数据。
• 查询员工信息 (GET /lst)：返回所有员工的详细信息。
• 修改员工信息 (PUT /upd)：根据 id 进行员工信息更新，支持部分字段修改。
• 删除员工信息 (DELETE /del?emp_id={id})：根据 id 删除对应的员工记录，并返回删除状态。

接口通过 依赖注入（Depends） 方式获取数据库会话，保证数据操作的一致性，同时返回的数据均为 JSON 格式，便于客户端解析。

import uvicorn
from fastapi import FastAPI, Depends
from sqlmodel import SQLModel, Field, create_engine, Session, select
from pydantic import BaseModel
import os
from typing import List


# 定义表结构
class Employee(SQLModel, table=True):
    id : int | None = Field(default=None, primary_key=True)
    name : str = Field(default='undefined')
    age : int = Field(default=0)
    salary : float = Field(default=0.0)


# 创建数据库
db_name = 'employee.db'
if os.path.exists(db_name):
    os.remove(db_name)

engine = create_engine(f'sqlite:///{db_name}')
SQLModel.metadata.create_all(engine)

# 创建 fastapi 应用
app = FastAPI()


# 创建依赖注入函数
def get_db_session():
    with Session(engine) as session:
        yield session


# 定义服务接口
# 1. 添加员工信息
@app.post('/add')
def add_emp(employee : Employee, session : Session = Depends(get_db_session)) -> Employee:
    # 数据入库
    session.add(employee)
    # 提交事务
    session.commit()
    # 更新数据
    session.refresh(employee)
    return employee


# 2. 展示员工信息
@app.get('/lst')
def lst_emp(session : Session = Depends(get_db_session)) -> List[Employee]:
    # 使用 select 构建一个 select 查询，并交给 session 去执行，通过 all 筛选出所有的数据，并返回
    # 注意: 执行结果是一个 List[Dict] 类型的数据
    employees = session.exec(select(Employee)).all()
    return employees


# 3. 删除员工信息
class DeleteResponse(BaseModel):
    status : str
    employee : Employee | None

@app.delete('/del')
def del_emp(emp_id : int, session : Session = Depends(get_db_session)) -> DeleteResponse:
    # 根据主键查询要删除的数据
    employee = session.get(Employee, emp_id)
    if not employee:
        return DeleteResponse(status='failure', employee=None)
    # 直接从数据库中删除指定数据
    session.delete(employee)
    session.commit()
    return DeleteResponse(status='success', employee=employee)


# 4. 修改员工信息
@app.put('/upd')
def upd_emp(employee : Employee, session : Session = Depends(get_db_session)) -> Employee:
    # 查询要修改的信息是否存在
    employee_db : Employee =  session.get(Employee, employee.id)
    # 将输入的 employee 对象转换为 dict 类型，并且将未设置字段删除
    employee_dict : dict = employee.model_dump(exclude_unset=True)
    # 更新 employee_db 对象
    employee_db.sqlmodel_update(employee_dict)
    # 将新的数据更新到数据库中
    session.add(employee_db)
    session.commit()
    session.refresh(employee_db)

    return employee_db


if __name__ == '__main__':
    uvicorn.run(app, host='127.0.0.1', port=8000)

2. 接口使用

案例中提供了 demo() 函数，使用 requests 库调用各个 API，完成以下操作：

1. 添加员工信息：调用 /add 接口，批量添加三名员工，并打印返回的员工信息。
2. 查询员工信息：调用 /lst 接口，获取所有员工的详细信息并打印。
3. 修改员工信息：调用 /upd 接口，修改某个员工的 name、age 和 salary，并查看更新后的数据。
4. 删除员工信息：调用 /del 接口，删除某个员工，并返回删除状态和被删除的员工数据

pip install requests

import requests


def demo():
    # 1. 添加员工信息
    employees = [{'id': None, 'name': 'obama', 'age': 70, 'salary': 5000.00},
                 {'id': None, 'name': 'smith', 'age': 40, 'salary': 3000.00},
                 {'id': None, 'name': 'polly', 'age': 90, 'salary': 7000.00}]

    for employee in employees:
        response = requests.post('http://127.0.0.1:8000/add', json=employee)
        print('添加的信息:', response.json())
    print('-' * 65)

    # 2. 列出员工信息
    response = requests.get('http://127.0.0.1:8000/lst')
    for employee in response.json():
        print('查询的数据:', employee)

    # 3. 修改员工信息
    update_employee = {'id': 2, 'name': 'new_name', 'age': 55, 'salary': 1000.00}
    response = requests.put('http://127.0.0.1:8000/upd', json=update_employee)
    print('修改的数据:', response.json())


    # 4. 删除员工信息
    params = {'emp_id': 1}
    response = requests.delete('http://127.0.0.1:8000/del', params=params)
    print('删除的数据:', response.json())


if __name__ == '__main__':
    demo()