在软件开发中，数据访问层（DAL）和业务逻辑层（BLL）之间的解耦是一个重要的设计原则。这种设计可以提高代码的可维护性和可扩展性。本文将介绍如何使用Python和SQLAlchemy创建一个通用的数据访问层，并展示如何在FastAPI项目中使用它。
通用数据访问层是一种设计模式，它在数据访问层和业务逻辑层之间创建了一个抽象层。这个抽象层的目的是隔离数据访问层，使得对数据访问层的更改不会直接影响业务逻辑层。通常，会使用通用的仓库类来避免代码的冗余。
在Python中，可以使用SQLAlchemy来创建通用仓库类。SQLAlchemy是一个流行的SQL工具包和对象关系映射（ORM）系统，它允许以面向对象的方式来操作数据库。
首先，需要定义一个基础模型，它包含了所有实体共有的字段。例如：
from sqlalchemy import Column, Integer, Boolean, DateTime
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class AppBaseModel:
__tablename__ = 'base_model'
id = Column(Integer, primary_key=True, index=True)
is_active = Column(Boolean, default=True)
created_by = Column(Integer)
updated_by = Column(Integer, default=None)
created_datetime = Column(DateTime, default=datetime.datetime.utcnow)
updated_datetime = Column(DateTime, default=None, onupdate=datetime.datetime.utcnow)
account_id = Column(Integer)
接下来，定义具体的数据表模型。例如，有两个数据表模型：TaskQueue和GroupQueue。
class TaskQueue(AppBaseModel, Base):
__tablename__ = 'task_queues'
name = Column(String, index=True)
class GroupQueue(AppBaseModel, Base):
__tablename__ = 'group_queues'
name = Column(String, index=True)
仓库类是基础的仓库类，它包含了一些基本的操作，如从表中读取数据、向表中添加/更新/删除行等。
from sqlalchemy.orm import Session
class TableRepository:
entity = None
db = None
def __init__(self, db: Session, entity):
self.db = db
self.entity = entity
def get_all(self):
return self.db.query(self.entity).all()
def get_by_id(self, id: int):
return self.db.query(self.entity).filter(self.entity.id == id).first()
# 更多方法...
在FastAPI项目中，可以直接使用TableRepository仓库类，而不需要任何继承。例如，对于TaskQueue实体和task_queues表，可以这样操作：
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
app = FastAPI()
@app.get("/task_queues/", response_model=List[TaskQueueSchema])
def get_all(task_queue_repo: TableRepository = Depends(lambda: TableRepository(get_db(), models.TaskQueue))):
return task_queue_repo.get_all()
有时候，可能需要对仓库类进行一些定制。在这种情况下，可以继承TableRepository类。例如，GroupQueueCrud类继承了TableRepository类，用于操作GroupQueue实体和group_queues表。
class GroupQueueCrud(TableRepository):
def __init__(self, db: Session):
super().__init__(db=db, entity=models.GroupQueue)
要运行这个项目，需要在后端文件夹中打开命令行，然后输入以下命令：
docker-compose up -d
项目将会在http://localhost:4003上运行。可以通过访问http://localhost:4003/docs来查看API文档。