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

前置环境

本项目基于 Python 生态，依赖包较少，无需复杂配置。

pip install fastapi uvicorn requests

SQLite3 是 Python 标准库的一部分，开箱即用。

技术选型

• FastAPI: 现代异步 Web 框架，自动生成交互式文档。
• SQLite3: 轻量级嵌入式数据库，适合本地或小型服务。
• Requests: 标准的 HTTP 请求库，方便调试接口。
• Uvicorn: 高性能 ASGI 服务器，驱动 FastAPI 运行。

数据库交互

操作 SQLite 遵循'连接 - 游标 - 执行 - 提交'的标准流程。

import sqlite3
conn = sqlite3.connect("test.db")
cursor = conn.cursor()

如果目标文件不存在，连接操作会自动创建它。执行完 SQL 语句后，务必提交变更并关闭连接。

# 创建表结构
cursor.execute('''CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT,
    age INTEGER
)''')

# 插入记录
cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", ("张三", 18))

# 查询数据
cursor.execute("SELECT * FROM users")
data = cursor.fetchall()

conn.commit()
conn.close()

接口定义

使用装饰器映射 URL 路径到 Python 函数。

from fastapi import FastAPI
app = FastAPI()

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

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

启动服务后，访问 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 文件。

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