快速上手：使用FastAPI与达梦数据库实现高效数据同步

在当今数据驱动的时代，如何快速构建高性能的API服务并与国产数据库无缝集成，成为了许多开发者的关注焦点。今天，我们将探讨如何使用FastAPI这一现代Python Web框架，结合达梦数据库（DM Database）实现高效的数据同步操作。

环境准备

1. 安装达梦数据库驱动

达梦数据库提供了专门的Python驱动程序dmPython，需要从达梦官网下载并安装：

# 进入达梦数据库安装目录下的python驱动文件夹
cd /opt/dmdbms/drivers/python/dmPython

# 执行安装命令
python setup.py install

2. 安装SQLAlchemy_dm方言包

为了支持SQLAlchemy操作达梦数据库，需要安装方言包：

# 进入达梦数据库安装目录下的SQLAlchemy_dm方言包文件夹
cd /opt/dmdbms/drivers/python/SQLAlchemy_dm

# 执行安装命令
python setup.py install

3. 安装FastAPI及其他依赖

1	pip install fastapi uvicorn sqlalchemy pydantic

核心代码实现

数据库连接配置

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# 创建同步数据库引擎
# 注意：这里使用了达梦数据库的连接字符串
engine = create_engine(
    "dm://SYSDBA:xxxxx@localhost:5236",
    echo=True  # 开启SQL语句日志，便于调试
)

# 创建同步会话工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 创建Base类
Base = declarative_base()

# 依赖注入：获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

数据模型定义

from sqlalchemy import Column, Integer, String

class Item(Base):
    __tablename__ = "items"
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(255), nullable=False)
    description = Column(String(500))
    price = Column(Integer)

Pydantic模型定义（数据验证）

from pydantic import BaseModel, ConfigDict
from typing import List

class ItemBase(BaseModel):
    name: str
    description: str = None
    price: int

class ItemCreate(ItemBase):
    pass

class ItemResponse(ItemBase):
    id: int
    model_config = ConfigDict(from_attributes=True)

FastAPI应用与API端点

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from sqlalchemy import select

app = FastAPI()

# 应用启动时创建数据库表
@app.on_event("startup")
def startup():
    Base.metadata.create_all(bind=engine)

# 创建项目
@app.post("/items/create_item", response_model=ItemResponse)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    try:
        db_item = Item(
            name=item.name, 
            description=item.description, 
            price=item.price
        )
        db.add(db_item)
        db.commit()
        db.refresh(db_item)
        return db_item
    except Exception as e:
        db.rollback()
        raise HTTPException(status_code=500, detail=f"创建失败: {str(e)}")

# 读取项目列表
@app.get("/items/read_items", response_model=List[ItemResponse])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    try:
        result = db.execute(select(Item).offset(skip).limit(limit))
        items = result.scalars().all()
        return items
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"读取失败: {str(e)}")

# 读取单个项目
@app.get("/items/read_item/{item_id}", response_model=ItemResponse)
def read_item(item_id: int, db: Session = Depends(get_db)):
    try:
        item = db.get(Item, item_id)
        if item is None:
            raise HTTPException(status_code=404, detail="项目不存在")
        return item
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"读取失败: {str(e)}")

# 更新项目
@app.put("/items/update_item/{item_id}", response_model=ItemResponse)
def update_item(item_id: int, item: ItemCreate, db: Session = Depends(get_db)):
    try:
        db_item = db.get(Item, item_id)
        if db_item is None:
            raise HTTPException(status_code=404, detail="项目不存在")
        
        db_item.name = item.name
        db_item.description = item.description
        db_item.price = item.price
        
        db.commit()
        db.refresh(db_item)
        return db_item
    except HTTPException:
        raise
    except Exception as e:
        db.rollback()
        raise HTTPException(status_code=500, detail=f"更新失败: {str(e)}")

# 删除项目
@app.delete("/items/delete_item/{item_id}")
def delete_item(item_id: int, db: Session = Depends(get_db)):
    try:
        db_item = db.get(Item, item_id)
        if db_item is None:
            raise HTTPException(status_code=404, detail="项目不存在")
        
        db.delete(db_item)
        db.commit()
        return {"message": "项目删除成功"}
    except HTTPException:
        raise
    except Exception as e:
        db.rollback()
        raise HTTPException(status_code=500, detail=f"删除失败: {str(e)}")

运行应用

1 2	# 启动FastAPI应用 uvicorn main:app --reload --host 0.0.0.0 --port 8000

应用启动后，可以通过以下地址访问API文档：

Swagger UI文档: http://127.0.0.1:8000/docs
ReDoc文档: http://127.0.0.1:8000/redoc

API测试示例

创建项目

1
2
3

curl -X POST "http://127.0.0.1:8000/items/create_item" \
     -H "Content-Type: application/json" \
     -d '{"name": "笔记本电脑", "description": "高性能游戏本", "price": 8999}'

查询项目列表

1	curl "http://127.0.0.1:8000/items/read_items?skip=0&limit=10"

查询单个项目

1	curl "http://127.0.0.1:8000/items/read_item/1"

更新项目

1
2
3

curl -X PUT "http://127.0.0.1:8000/items/update_item/1" \
     -H "Content-Type: application/json" \
     -d '{"name": "游戏笔记本", "description": "RTX 4060显卡", "price": 9999}'