正文

掌握FastAPI，轻松实现数据库交互实战攻略

/0 浏览量
1212

简介

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于构建 API，由 Python 3.6+ 支持。它基于标准 Python 类型提示，具有内置的自动文档生成和验证功能。在本文中，我们将探讨如何使用 FastAPI 与数据库进行交互，包括创建、读取、更新和删除（CRUD）操作。

环境准备

在开始之前，请确保您已经安装了以下软件：

  • Python 3.6 或更高版本
  • FastAPI
  • Uvicorn（作为 ASGI 服务器）
  • 数据库（例如 PostgreSQL、MySQL、SQLite）

您可以使用以下命令安装所需的包：

pip install fastapi uvicorn databases sqlalchemy

数据库选择与配置

选择一个数据库系统，并配置您的数据库连接。以下是一个使用 SQLite 数据库的例子：

from databases import Database
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String

DATABASE_URL = "sqlite:///./test.db"

database = Database(DATABASE_URL)

metadata = MetaData()

users = Table(
    "users", metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String),
    Column("email", String),
)

engine = create_engine(DATABASE_URL)
metadata.create_all(engine)

创建模型

定义您的数据库模型，使用 SQLAlchemy 的 ORM 功能。

from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str
    email: str

创建 FastAPI 应用

创建一个 FastAPI 应用，并定义路由。

from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

实现数据库交互

创建用户

@app.post("/users/")
async def create_user(user: User):
    query = users.insert().values(name=user.name, email=user.email)
    last_record_id = await database.execute(query)
    return {**user.dict(), "id": last_record_id}

获取用户列表

@app.get("/users/")
async def get_users():
    query = users.select()
    results = await database.fetch_all(query)
    return [{"id": result["id"], "name": result["name"], "email": result["email"]} for result in results]

获取单个用户

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    query = users.select().where(users.c.id == user_id)
    result = await database.fetch_one(query)
    if result:
        return {"id": result["id"], "name": result["name"], "email": result["email"]}
    else:
        return {"message": "User not found"}

更新用户

@app.put("/users/{user_id}")
async def update_user(user_id: int, user: User):
    query = users.update().values(name=user.name, email=user.email).where(users.c.id == user_id)
    result = await database.execute(query)
    if result == 1:
        return {"id": user_id, "name": user.name, "email": user.email}
    else:
        return {"message": "User not found"}

删除用户

@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
    query = users.delete().where(users.c.id == user_id)
    result = await database.execute(query)
    if result == 1:
        return {"message": "User deleted"}
    else:
        return {"message": "User not found"}

运行应用

使用 Uvicorn 运行您的 FastAPI 应用：

uvicorn your_module:app --reload

替换 your_module 为包含您的 FastAPI 应用的 Python 模块名称。

总结

本文介绍了如何使用 FastAPI 和 SQLAlchemy 与数据库进行交互。通过以上步骤，您应该能够创建一个基本的 API 来管理用户数据。FastAPI 的自动文档生成功能将帮助您快速了解如何使用您的 API。随着您对 FastAPI 和数据库交互的深入了解，您可以扩展这些概念以支持更复杂的业务逻辑。

-- 展开阅读全文 --