DDD分层设计与异步职责划分：让你的代码不再“异步”混乱
title: DDD分层设计与异步职责划分：让你的代码不再“异步”混乱
# 同步领域服务
class CommentService:
def validate_comment(self, content: str):
"""领域层保持同步逻辑"""
if len(content) > 1000:
raise ValueError("评论内容过长")
# 其他业务规则...
# 异步基础设施
class CommentRepository:
async def save(self, comment):
"""基础设施层处理异步IO"""
await comment.save()
from pydantic import BaseModel, Field
from datetime import datetime
class User(BaseModel):
id: int
username: str = Field(..., max_length=50)
class Comment(BaseModel):
content: str = Field(..., min_length=1, max_length=1000)
author: User
created_at: datetime = datetime.now()
def edit(self, new_content: str):
"""领域方法保持同步"""
if len(new_content) > 1000:
raise ValueError("内容超过最大限制")
self.content = new_content
self.updated_at = datetime.now()
from tortoise.models import Model
from tortoise import fields
class CommentDBModel(Model):
"""数据库模型"""
id = fields.IntField(pk=True)
content = fields.TextField()
author_id = fields.IntField()
created_at = fields.DatetimeField(auto_now_add=True)
class TortoiseCommentRepository:
async def save(self, comment: Comment) -> int:
"""异步保存方法"""
db_model = await CommentDBModel.create(
content=comment.content,
author_id=comment.author.id
)
return db_model.id
async def get(self, comment_id: int) -> Comment:
"""异步查询方法"""
db_model = await CommentDBModel.get(id=comment_id)
return Comment(
content=db_model.content,
author=User(id=db_model.author_id, username=""),
created_at=db_model.created_at
)
from fastapi import APIRouter, Depends
router = APIRouter()
@router.post("/comments")
async def create_comment(
comment: Comment,
repo: TortoiseCommentRepository = Depends()
):
# 调用同步领域验证
CommentService().validate_comment(comment.content)
# 异步保存操作
comment_id = await repo.save(comment)
return {"id": comment_id}
HTTP请求 -> 路由层（async）-> 应用层（async）-> 领域层（sync）-> 基础设施层（async）
from tortoise.transactions import atomic
class CommentRepository:
@atomic()
async def create_with_user(self, comment: Comment, user: User):
"""事务操作示例"""
await UserDBModel.create(id=user.id, username=user.username)
await CommentDBModel.create(
content=comment.content,
author_id=user.id
)
# 环境要求
pip install fastapi uvicorn tortoise-orm pydantic
# 启动命令
uvicorn main:app --reload
评论
发表评论