【FastAPI】使用 SQLAlchemy 和 FastAPI 实现 PostgreSQL 中的 JSON 数据 CRUD 操作

在现代 web 开发中，处理 JSON 数据变得越来越普遍。本文将指导你如何使用 FastAPI 和 SQLAlchemy 实现对 PostgreSQL 数据库中 JSON 数据的增删改查（CRUD）操作。

环境准备

首先，确保你已经安装了所需的库。在终端中运行以下命令：

pip install fastapi[all] sqlalchemy psycopg2

这些库分别用于构建 API、与数据库交互以及 PostgreSQL 的连接。

项目结构

在开始之前，建议按照以下结构组织你的项目：

my_project/
│
├── main.py        # FastAPI 应用
└── models.py      # 数据库模型

1. 创建数据库模型

在 models.py 中定义我们的数据库模型。我们将使用 SQLAlchemy 来创建与 PostgreSQL 的连接。

from sqlalchemy import create_engine, Column, Integer, String, JSON
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql://user:password@localhost/dbname"

Base = declarative_base()
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

class Item(Base):
    __tablename__ = 'items'
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    data = Column(JSON)

# 创建表
Base.metadata.create_all(bind=engine)

在这里，我们定义了一个 Item 类，包含 id、name 和 data 字段，其中 data 字段将用于存储 JSON 数据。

2. 创建 FastAPI 应用

接下来，在 main.py 中构建 FastAPI 应用并实现 CRUD 操作。

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from models import SessionLocal, Item

app = FastAPI()

# 创建数据库会话依赖
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 创建项目
@app.post("/items/", response_model=Item)
def create_item(item: Item, db: Session = Depends(get_db)):
    db.add(item)
    db.commit()
    db.refresh(item)
    return item

# 获取项目
@app.get("/items/{item_id}", response_model=Item)
def read_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(Item).filter(Item.id == item_id).first()
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

# 更新项目
@app.put("/items/{item_id}", response_model=Item)
def update_item(item_id: int, item: Item, db: Session = Depends(get_db)):
    db_item = db.query(Item).filter(Item.id == item_id).first()
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    
    db_item.name = item.name
    db_item.data = item.data
    db.commit()
    db.refresh(db_item)
    return db_item

# 删除项目
@app.delete("/items/{item_id}", response_model=Item)
def delete_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(Item).filter(Item.id == item_id).first()
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    
    db.delete(item)
    db.commit()
    return item

在这个应用中，我们定义了四个主要的路由，分别用于创建、读取、更新和删除项目。

3. 运行应用

在终端中运行以下命令启动 FastAPI 应用：

uvicorn main:app --reload

应用启动后，你可以访问 http://127.0.0.1:8000/docs 查看自动生成的 API 文档，并进行测试。

4. 测试 API

你可以使用 Postman 或 curl 测试 API 操作，例如：

• 创建项目：

curl -X POST "http://127.0.0.1:8000/items/" -H "Content-Type: application/json" -d '{"name": "item1", "data": {"key": "value"}}'

• 获取项目：

curl -X GET "http://127.0.0.1:8000/items/1"

• 更新项目：

curl -X PUT "http://127.0.0.1:8000/items/1" -H "Content-Type: application/json" -d '{"name": "item1_updated", "data": {"key": "new_value"}}'

• 删除项目：

curl -X DELETE "http://127.0.0.1:8000/items/1"

结论

通过以上步骤，你可以使用 FastAPI 和 SQLAlchemy 实现对 PostgreSQL 中 JSON 数据的增删改查操作。这种组合不仅能提供高性能的 API 体验，还能方便地处理复杂的数据结构。

希望这篇博客对你在项目中实现 CRUD 操作有所帮助！如果你有任何问题或想法，欢迎在评论区留言讨论。