引言
FastAPI 是一个现代、快速(高性能)的 Web 框架,用于构建 API,它基于标准 Python 类型提示。在本文中,我们将深入探讨如何使用 FastAPI 与数据库进行高效交互,并通过一个实战案例来展示这一过程。
FastAPI 简介
FastAPI 是一个高性能的 Web 框架,它利用了 Python 3.6+ 的类型提示功能,以及 Starlette 和 Pydantic。它具有以下特点:
- 高性能:FastAPI 使用异步编程,能够处理大量的并发请求。
- 易于使用:FastAPI 的 API 定义非常直观,易于理解和编写。
- 自动文档:FastAPI 可以自动生成交互式 API 文档。
数据库交互概述
在 FastAPI 中,数据库交互通常涉及以下步骤:
- 选择数据库。
- 创建数据库连接。
- 定义模型。
- 创建数据库操作函数。
实战案例:使用 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 数据库交互的潜力。
