正文

解锁FastAPI高效数据库交互：实战案例带你轻松入门

/0 浏览量
1128

引言

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于构建 API，由 Python 3.6+ 类型提示驱动。它具有简洁的 API 设计，支持异步请求处理，并且可以与多种数据库进行高效交互。本文将带你通过实战案例，轻松入门 FastAPI 与数据库的交互。

准备工作

在开始之前，请确保你已经安装了以下工具和库：

  • Python 3.6 或更高版本
  • FastAPI
  • Uvicorn（用于运行 FastAPI 应用）
  • Pydantic（用于数据验证和序列化）
  • SQL 数据库（如 SQLite、PostgreSQL 等）

创建 FastAPI 应用

首先，创建一个新的 Python 项目，并在其中安装 FastAPI 和 Uvicorn：

pip install fastapi uvicorn

然后，创建一个名为 main.py 的文件，并编写以下代码来初始化 FastAPI 应用：

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Welcome to FastAPI database interaction tutorial!"}

保存并运行应用：

uvicorn main:app --reload

访问 http://127.0.0.1:8000/，你应该会看到一个欢迎消息。

数据库配置

选择一个数据库并创建一个数据库实例。以下是使用 SQLite 的示例：

import sqlite3

DATABASE_URL = "sqlite:///./test.db"
database = sqlite3.connect(DATABASE_URL)

定义模型

使用 Pydantic 定义数据库模型：

from pydantic import BaseModel

class Item(BaseModel):
    id: int
    name: str
    description: str = None
    price: float
    tax: float = None

创建数据库表

使用 SQL 语句创建数据库表：

import sqlalchemy

metadata = sqlalchemy.MetaData()

items = sqlalchemy.Table(
    "items", metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True, index=True),
    sqlalchemy.Column("name", sqlalchemy.String(80), nullable=False),
    sqlalchemy.Column("description", sqlalchemy.String(255), nullable=True),
    sqlalchemy.Column("price", sqlalchemy.Float, nullable=False),
    sqlalchemy.Column("tax", sqlalchemy.Float, nullable=True),
)

metadata.create_all(database.engine)

数据库交互

现在我们可以使用 FastAPI 和 SQLAlchemy 进行数据库交互了。以下是一些基本的 CRUD（创建、读取、更新、删除）操作：

创建

from sqlalchemy.orm import Session

def create_item(item: Item):
    db = Session()
    db.add(item)
    db.commit()
    db.refresh(item)
    return item

读取

def read_item(item_id: int):
    db = Session()
    item = db.query(Item).filter(Item.id == item_id).first()
    return item

更新

def update_item(item_id: int, item: Item):
    db = Session()
    db_item = db.query(Item).filter(Item.id == item_id).first()
    db_item.name = item.name
    db_item.description = item.description
    db_item.price = item.price
    db_item.tax = item.tax
    db.commit()
    db.refresh(db_item)
    return db_item

删除

def delete_item(item_id: int):
    db = Session()
    db_item = db.query(Item).filter(Item.id == item_id).first()
    db.delete(db_item)
    db.commit()
    return db_item

实战案例

现在，让我们通过一个简单的 API 来展示如何使用 FastAPI 和数据库进行交互。我们将创建一个 API 来处理商品项。

创建 API

在 main.py 中添加以下路由：

from fastapi import HTTPException

@app.post("/items/")
async def create_item(item: Item):
    return create_item(item)

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    item = read_item(item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    updated_item = update_item(item_id, item)
    return updated_item

@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    deleted_item = delete_item(item_id)
    return deleted_item

运行应用

再次运行应用，并使用 Postman 或任何 API 测试工具来测试这些路由。

总结

通过本文的实战案例，你现在已经了解了如何使用 FastAPI 与数据库进行高效交互。FastAPI 的异步处理能力和类型提示功能使得构建高性能的 Web API 变得更加简单和愉快。希望这个教程能够帮助你快速入门 FastAPI 数据库交互。

-- 展开阅读全文 --