Python 入门：FastAPI + SQLite3 + Requests 基础教学 | 极客日志

Python 入门：FastAPI + SQLite3 + Requests 基础教学 | 极客日志

import sqlite3

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

cursor = conn.cursor()

# 创建表
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()    # 关闭连接

from fastapi import FastAPI
app = FastAPI()

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

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

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

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

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

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

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

# 1. 添加用户
def add_user():
    res = requests.post(f"{url}/add_user", params={
        "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()