Python 实战：FastAPI + SQLite3 + Requests 接口开发教程

启动服务后，访问 http://127.0.0.1:8000/docs 即可查看 Swagger UI 文档。

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

项目整合

将数据库逻辑封装进 API 服务中。新建 main.py，在应用启动时初始化表结构。

import sqlite3
from fastapi import FastAPI
import uvicorn

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

@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

@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

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

注意：获取列表接口已修正返回值，确保能正确返回用户数据而非固定布尔值。

接口测试

使用 requests 模拟客户端请求，验证接口连通性。

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

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

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

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

运行步骤

1. 启动 main.py 开启服务。
2. 运行测试脚本发送请求。
3. 观察控制台输出及生成的 data.db 文件。

整个流程涵盖了从数据库初始化到接口调用的闭环，掌握了这些基础，后续可扩展更多业务逻辑。