扫描二维码
关注或者微信搜一搜：编程智域 前端至全栈交流与成长
发现1000+提升效率与开发的AI工具和实用程序：https://tools.cmdragon.cn/
# 安装依赖：sqlalchemy==1.4.46
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(50))
email = Column(String(120), unique=True)
posts = relationship("Post", back_populates="author")
class Post(Base):
__tablename__ = 'posts'
id = Column(Integer, primary_key=True)
title = Column(String(100))
content = Column(String(500))
author_id = Column(Integer, ForeignKey('users.id'))
author = relationship("User", back_populates="posts")
1.2 FastAPI集成配置
使用依赖注入实现数据库会话管理：
# 安装依赖：databases==0.7.0
from fastapi import Depends
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/users/{user_id}")
async def read_user(user_id: int, db: Session = Depends(get_db)):
user = db.query(User).filter(User.id == user_id).first()
return user
# 安装依赖：strawberry-graphql==0.187.0
import strawberry
from typing import List
@strawberry.type
class PostType:
id: int
title: str
content: str
@strawberry.type
class UserType:
id: int
name: str
email: str
posts: List[PostType]
@strawberry.type
class Query:
@strawberry.field
async def users(self, info) -> List[UserType]:
db = info.context["db"]
return db.query(User).all()
schema = strawberry.Schema(query=Query)
2.2 自动类型转换机制
创建Pydantic模型实现数据验证：
# 安装依赖：pydantic==1.10.7
from pydantic import BaseModel
class UserCreate(BaseModel):
name: str
email: str
class UserResponse(UserCreate):
id: int
@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
db_user = User(**user.dict())
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
class UserService:
@staticmethod
def get_users(db: Session):
return db.query(User).all()
@staticmethod
def create_user(db: Session, user_data: dict):
user = User(**user_data)
db.add(user)
db.commit()
return user
@strawberry.type
class Mutation:
@strawberry.mutation
def create_user(self, info, name: str, email: str) -> UserType:
db = info.context["db"]
user_data = {"name": name, "email": email}
return UserService.create_user(db, user_data)
{
"detail": [
{
"loc": [
"body",
"email"
],
"msg": "value is not a valid email address",
"type": "value_error.email"
}
]
}
解决方法：
4.2 数据库连接错误
配置检查清单：
Q2：如何保证GraphQL接口的安全性？
A) 禁用所有查询参数
B) 实现查询深度限制
C) 开放所有字段查询
正确答案：B。使用最大查询深度限制和查询成本分析来防止恶意复杂查询，同时配合JWT认证。
Q3：当需要同时支持REST和GraphQL接口时，最佳实践是：
A) 分别开发独立接口
B) 使用Schema转换工具
C) 基于服务层构建业务逻辑
正确答案：C。通过抽象服务层实现业务逻辑复用，使不同接口协议共享相同核心逻辑。
余下文章内容请点击跳转至 个人博客页面 或者 扫码关注或者微信搜一搜：编程智域 前端至全栈交流与成长
，阅读完整的文章：SQLAlchemy ORM与GraphQL的完美邂逅，如何让数据库操作变得如此简单？
登录查看全部
参与评论
手机查看
返回顶部