Python 实战：FastAPI + SQLite3 + Requests 基础开发

@app.get("/hello")
def hello():
    return {"message": "Hello FastAPI"}

3.3 定义带参数 GET 接口

@app.get("/user")
def get_user(name: str):
    return {"你的名字": name}

3.4 定义 POST 接口

@app.post("/add")
def add(name: str, age: int):
    return {"接收数据": {"name": name, "age": age}}

3.5 运行服务

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

可以通过 http://127.0.0.1:8000/docs 来访问查看接口和调试。

四、完整整合 FastAPI + SQLite3

新建文件 main.py，以下为可直接运行的完整代码。

import sqlite3
from fastapi import FastAPI
import uvicorn

# 创建 FastAPI 实例
app = FastAPI()

# 初始化数据库
def init_db():
    conn = sqlite3.connect("data.db")
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            age INTEGER NOT NULL
        )
    ''')
    conn.commit()
    conn.close()

# 程序启动时初始化
init_db()

# 接口 1：添加用户（POST）
@app.post("/api/add_user")
def add_user(name: str, age: int):
    try:
        conn = sqlite3.connect("data.db")
        cursor = conn.cursor()
        cursor.execute("INSERT INTO users (name, age) VALUES (?,?)", (name, age))
        conn.commit()
        conn.close()
        return True
    except Exception as e:
        return False

# 接口 2：获取所有用户（GET）
@app.get("/api/get_users")
def get_users():
    conn = sqlite3.connect("data.db")
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    users = cursor.fetchall()
    conn.close()
    result = []
    for u in users:
        result.append({
            "id": u[0],
            "name": u[1],
            "age": u[2]
        })
    return result  # 修正：原代码返回 True，此处应返回实际数据

# 运行服务
if __name__ == "__main__":
    uvicorn.run("main:app", host="127.0.0.1", port=8000)

五、使用 Requests 请求接口（测试）

新建 test_request.py，用于调用上面的 API。

import requests

url = "http://127.0.0.1:8000/api"

# 1. 添加用户
def add_user():
    res = requests.post(f"{url}/add_user", json={
        "name": "Python 学习者",
        "age": 20
    })
    print("添加结果：", res.json())

# 2. 获取用户列表
def get_users():
    res = requests.get(f"{url}/get_users")
    print("所有用户：", res.json())

if __name__ == "__main__":
    add_user()
    get_users()

注意：POST 请求建议使用 json= 参数传递数据体，以便 FastAPI 正确解析。

六、运行步骤

1. 运行 main.py 启动 API 服务。
2. 运行 test_request.py 发送请求。
3. 查看控制台输出。
4. 数据库文件 data.db 自动生成。

七、核心逻辑总结

1. SQLite 流程：连接 → 创建游标 → 执行 SQL → 提交 → 关闭。
2. FastAPI 流程：创建 app → 装饰器定义接口 → 编写函数 → 运行服务。
3. 传参方式：GET/POST 均可用，但 POST 建议用 json= 传参。
4. Requests 流程：指定 URL → 传参 → 获取响应 → 解析 JSON。