正文

解锁FastAPI高效数据库交互：实战案例带你飞

/0 浏览量
1128

引言

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于构建 API，它基于标准 Python 类型提示。在本文中，我们将深入探讨如何使用 FastAPI 与数据库进行高效交互，并通过一个实战案例来展示这一过程。

FastAPI 简介

FastAPI 是一个高性能的 Web 框架，它利用了 Python 3.6+ 的类型提示功能，以及 Starlette 和 Pydantic。它具有以下特点：

  • 高性能：FastAPI 使用异步编程，能够处理大量的并发请求。
  • 易于使用：FastAPI 的 API 定义非常直观，易于理解和编写。
  • 自动文档：FastAPI 可以自动生成交互式 API 文档。

数据库交互概述

在 FastAPI 中，数据库交互通常涉及以下步骤：

  1. 选择数据库。
  2. 创建数据库连接。
  3. 定义模型。
  4. 创建数据库操作函数。

实战案例：使用 SQLite 与 FastAPI 交互

在这个案例中，我们将使用 SQLite 数据库和 FastAPI 来创建一个简单的博客系统。

1. 安装依赖

首先，我们需要安装 FastAPI 和 Uvicorn，以及用于数据库交互的 sqlalchemy 和 pydantic：

pip install fastapi uvicorn sqlalchemy pydantic

2. 创建数据库模型

使用 Pydantic 定义数据模型：

from pydantic import BaseModel
from typing import Optional

class Blog(BaseModel):
    title: str
    content: str
    published: Optional[bool] = True

3. 创建数据库连接

使用 SQLAlchemy 创建数据库连接：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

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

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

4. 创建数据库操作函数

创建函数来添加、更新、删除和获取博客条目：

def create_blog(title: str, content: str, published: bool = True):
    db = get_db()
    new_blog = Blog(title=title, content=content, published=published)
    db.add(new_blog)
    db.commit()
    db.refresh(new_blog)
    return new_blog

def get_blog(id: int):
    db = get_db()
    blog = db.query(Blog).filter(Blog.id == id).first()
    return blog

def update_blog(id: int, title: str, content: str, published: bool):
    db = get_db()
    blog = db.query(Blog).filter(Blog.id == id).first()
    blog.title = title
    blog.content = content
    blog.published = published
    db.commit()
    db.refresh(blog)
    return blog

def delete_blog(id: int):
    db = get_db()
    blog = db.query(Blog).filter(Blog.id == id).first()
    db.delete(blog)
    db.commit()

5. 创建 FastAPI 应用

创建 FastAPI 应用，并定义路由：

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.post("/blogs/")
def create_blog(title: str, content: str, published: bool = True):
    return create_blog(title, content, published)

@app.get("/blogs/{id}")
def read_blog(id: int):
    blog = get_blog(id)
    if not blog:
        raise HTTPException(status_code=404, detail="Blog not found")
    return blog

@app.put("/blogs/{id}")
def update_blog(id: int, title: str, content: str, published: bool):
    blog = update_blog(id, title, content, published)
    return blog

@app.delete("/blogs/{id}")
def delete_blog(id: int):
    blog = delete_blog(id)
    return {"detail": "Deleted"}

6. 运行应用

使用 Uvicorn 运行 FastAPI 应用：

uvicorn main:app --reload

现在，你可以通过浏览器或 API 工具（如 Postman）来测试这些端点了。

总结

通过本文，我们学习了如何使用 FastAPI 与 SQLite 数据库进行高效交互。这个实战案例展示了如何定义数据模型、创建数据库连接、编写数据库操作函数，并最终将它们集成到 FastAPI 应用中。希望这个案例能够帮助你解锁 FastAPI 数据库交互的潜力。

-- 展开阅读全文 --