ToB企服应用市场:ToB评测及商务社交产业平台

标题: Python开发中，SQLAlchemy 的同步操作和异步操作封装，以及常规CRUD的处理 [打印本页]

---

作者: 用户国营    时间: 2024-8-22 09:54
标题: Python开发中，SQLAlchemy 的同步操作和异步操作封装，以及常规CRUD的处理
在我们使用Python来和数据库打交道中，SQLAlchemy是一个非常不错的ORM工具，通过它我们可以很好的实现多种数据库的统一模子接入，而且它提供了非常多的特性，通过团结不同的数据库驱动，我们可以实现同步或者异步的处理封装。1、SQLAlchemy介绍

SQLAlchemy 是一个功能强盛且机动的 Python SQL 工具包和对象关系映射（ORM）库。它被广泛用于在 Python 项目中处理关系型数据库的场景，既提供了高级的 ORM 功能，又保留了对底层 SQL 语句的强盛控制力。SQLAlchemy 允许开发者通过 Python 代码与数据库进行交互，而无需直接编写 SQL 语句，同时也支持直接使用原生 SQL 进行复杂查询。下面是SQLAlchemy和我们常规数据库对象的对应关系说明。Engine  　　连接对象         驱动引擎Session 　　连接池           事件  由此开始查询Model   　　表                   类定义Column  　  列  Query   　　若干行         可以链式添加多个条件 在使用SQLAlchemy时，通常会将其与数据库对象对应起来。以下是SQLAlchemy和常规数据库对象的对应关系说明：1）数据库表 (Database Table)

• SQLAlchemy: 使用 Table 对象或 Declarative Base 中的类来表示。
  
• 对应关系: 数据库中的每一个表对应于SQLAlchemy中的一个类，该类继承自 declarative_base()。
  

1. from sqlalchemy import Column, Integer, String, create_engine
2. from sqlalchemy.ext.declarative import declarative_base
4. Base = declarative_base()
6. class User(Base):
7.     __tablename__ = 'users'  # 数据库表名
8.     id = Column(Integer, primary_key=True)
9.     name = Column(String)
10.     email = Column(String)

复制代码

2）数据库列 (Database Column)

• SQLAlchemy: 使用 Column 对象来表示。
  
• 对应关系: 每个数据库表中的列在SQLAlchemy中表示为 Column 对象，并作为类的属性定义。
  

1. id = Column(Integer, primary_key=True)
2. name = Column(String(50))

复制代码

3）数据库行 (Database Row)

• SQLAlchemy: 每个数据库表的一个实例（对象）代表数据库表中的一行。
  
• 对应关系: 在SQLAlchemy中，通过实例化模子类来表示数据库表中的一行。
  

1. new_user = User(id=1, name='John Doe', email='john@example.com')

复制代码

4）主键 (Primary Key)

• SQLAlchemy: 使用 primary_key=True 参数定义主键。
  
• 对应关系: 在数据库表中定义主键列，这列在SQLAlchemy中也需要明确标注。
  

1. id = Column(Integer, primary_key=True)

复制代码

5）外键 (Foreign Key)

• SQLAlchemy: 使用 ForeignKey 对象来表示。
  
• 对应关系: 在SQLAlchemy中使用 ForeignKey 指定关系，指向另一个表的主键列。
  

1. from sqlalchemy import ForeignKey
2. from sqlalchemy.orm import relationship
4. class Address(Base):
5.     __tablename__ = 'addresses'
6.     id = Column(Integer, primary_key=True)
7.     user_id = Column(Integer, ForeignKey('users.id'))
8.     user = relationship('User')

复制代码

6）关系 (Relationships)

• SQLAlchemy: 使用 relationship 对象来表示。
  
• 对应关系: 数据库中表与表之间的关系在SQLAlchemy中通过 relationship 来定义。
  

1. addresses = relationship("Address", back_populates="user")

复制代码

7）会话 (Session)

• SQLAlchemy: 使用 Session 对象进行事件性操作（如查询、插入、更新、删除）。
  
• 对应关系: Session 对象雷同于数据库连接对象，用于与数据库进行交互。
  

1. from sqlalchemy.orm import sessionmaker
3. Session = sessionmaker(bind=engine)
4. session = Session()
6. session.add(new_user)
7. session.commit()

复制代码

通过以上对应关系，SQLAlchemy允许开发者以面向对象的方式与数据库交互，提供了一个Pythonic的接口来操作数据库。
 2、SQLAlchemy 的同步操作

 SQLAlchemy 提供了同步和异步两种操作方式，分别适用于不同的应用场景。以下是如何封装 SQLAlchemy 的同步和异步操作的方法说明：
在同步操作中，SQLAlchemy 使用传统的阻塞方式进行数据库操作。起首，定义一个底子的 Session 和 Engine 对象：

1. from sqlalchemy import create_engine
2. from sqlalchemy.orm import declarative_base, sessionmaker
3. from typing import Generator
4. from core.config import settings
6. # 常规同步处理
7. engine = create_engine(settings.DB_URI)
8. SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
10. def get_db() -> Generator:
11.     """创建一个 SQLAlchemy 数据库会话-同步处理."""
12.     try:
13.         db = SessionLocal()
14.         yield db
15.     finally:
16.         db.close()

复制代码

前面说了，使用SQLAlchemy可以实现不同数据库的统一模子的处理，我们可以对应创建不同数据库的连接（engine），如下是常规几种关系型数据库的连接处理。

1. # mysql 数据库引擎
2. engine = create_engine(
3.     "mysql+pymysql://root:123456@127.0.0.1:3306/WinFramework",
4.     pool_recycle=3600,
5.     # echo=True,
6. )
8. # Sqlite 数据库引擎
9. engine = create_engine("sqlite:///testdir//test.db")
11. # PostgreSQL 数据库引擎
12. engine = create_engine(
13.     "postgresql+psycopg2://postgres:123456@localhost:5432/winframework",
14.      # echo=True,
15. )
17. # SQLServer 数据库引擎
18. engine = create_engine(
19.      "mssql+pymssql://sa:123456@localhost/WinFramework?tds_version=7.0",
20.      # echo=True,
21. )

复制代码

我们可以根据数据库的CRUD操作方式，封装一些操作，如下所示。

1. class CRUDOperations:
2.     def __init__(self, model):
3.         self.model = model
5.     def create(self, db, obj_in):
6.         db_obj = self.model(**obj_in.dict())
7.         db.add(db_obj)
8.         db.commit()
9.         db.refresh(db_obj)
10.         return db_obj
12.     def get(self, db, id):
13.         return db.query(self.model).filter(self.model.id == id).first()
15.     def update(self, db, db_obj, obj_in):
16.         obj_data = obj_in.dict(exclude_unset=True)
17.         for field in obj_data:
18.             setattr(db_obj, field, obj_data[field])
19.         db.commit()
20.         db.refresh(db_obj)
21.         return db_obj
23.     def remove(self, db, id):
24.         obj = db.query(self.model).get(id)
25.         db.delete(obj)
26.         db.commit()
27.         return obj

复制代码

使用时，构建数据访问类进行操作，如下测试代码所示。

1. crud_user =<strong> CRUDOperations</strong>(<strong>User</strong>)
3. # Create
4. with get_db() as db:
5.     user = crud_user.create(db, user_data)
7. # Read
8. with get_db() as db:
9.     user = crud_user.get(db, user_id)
11. # Update
12. with get_db() as db:
13.     user = crud_user.update(db, user, user_data)
15. # Delete
16. with get_db() as db:
17.     crud_user.remove(db, user_id)

复制代码

 
3、SQLAlchemy 的异步操作封装

对于异步操作，SQLAlchemy 使用 AsyncSession 来管理异步事件。
起首，定义一个异步的 Session 和 Engine 对象：

1. from sqlalchemy import create_engine, URL
2. from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
3. from typing import AsyncGenerator
5. def create_engine_and_session(url: str | URL):
6.     try:
7.         # 数据库引擎
8.         engine = create_async_engine(url, pool_pre_ping=True)
9.     except Exception as e:
10.         print("❌ 数据库链接失败 {}", e)
11.         sys.exit()
12.     else:
13.         db_session = async_sessionmaker(
14.             bind=engine, autoflush=False, expire_on_commit=False
15.         )
16.         return engine, db_session
19. # 异步处理
20. async_engine, async_session = create_engine_and_session(settings.DB_URI_ASYNC)
23. async def get_db() -> AsyncGenerator[AsyncSession, None]:
24.     """创建一个 SQLAlchemy 数据库会话-异步处理."""
25.     async with async_session() as session:
26.         yield session

复制代码

和同步的处理雷同，不过是换了一个对象来实现，并且函数使用了async await的组合来实现异步操作。
为了实现我的SQLSugar开发框架雷同的封装模式，我们参考SQLSugar开发框架中基类CRUD的定义方式来实现多种接口的封装处理。

 参照上面的实现方式，我们来看看Python中使用泛型的处理封装类的代码。

1. ModelType = TypeVar("ModelType", bound=Base)
2. PrimaryKeyType = TypeVar("PrimaryKeyType", int, str, float)  # 限定主键的类型
3. PageDtoType = TypeVar("PageDtoType", bound=BaseModel)
4. DtoType = TypeVar("DtoType", bound=BaseModel)
7. class BaseCrud(Generic[ModelType, PrimaryKeyType, PageDtoType, DtoType]):
8.     """
9.     基础CRUD操作类
10.     """
12.     def __init__(self, model: Type[ModelType]):
13.         """
14.         数据库访问操作的基类对象(CRUD).
15.         **Parameters**
16.         * `model`: A SQLAlchemy model class
17.         """
18.         self.model = model

复制代码

这样，我们就可以通过泛型定义不同的类型，以及相关的处理类的信息。
该基类函数中，异步定义get_all的返回所有的数据接口如下所示。

1.     async def get_all(
2.         self, sorting: Optional[str], db: AsyncSession
3.     ) -> List[ModelType] | None:
4.         """根据ID字符串列表获取对象列表
6.         :param sorting: 格式：name asc 或 name asc,age desc
7.         """
8.         query = select(self.model)
9.         if sorting:
10.             query = self.apply_sorting(query, sorting)
12.         result = await db.execute(query)
13.         items = result.scalars().all()
14.         return items

复制代码

而对应获得单个对象的操作函数，如下所示。

1.     async def get(self, id: PrimaryKeyType, db: AsyncSession) -> Optional[ModelType]:
2.         """根据主键获取一个对象"""
3.         query = select(self.model).filter(self.model.id == id)
5.         result = await db.execute(query)
6.         item = result.scalars().first()
8.         return item

复制代码

而创建对象的操作函数，如下所示。

1.     async def create(self, obj_in: DtoType, db: AsyncSession, **kwargs) -> bool:
2.         """创建对象,使用 kwargs 时可以扩展创建对象时的字段。
4.         :param obj_in: 对象输入数据
5.         :param kwargs: 扩展字段,如格式: is_deleted=0, is_active=1
6.         """
7.         try:
8.             if kwargs:
9.                 instance = self.model(**obj_in.model_dump(), **kwargs)
10.             else:
11.                 instance = self.model(**obj_in.model_dump())  # type: ignore
13.             db.add(instance)
14.             await db.commit()
15.             return True
16.         except SQLAlchemyError as e:
17.             print(e)
18.             await db.rollback()
19.             return False

复制代码

这个异步函数 create 旨在通过 SQLAlchemy 在数据库中创建一个对象，同时允许通过 kwargs 参数动态扩展创建对象时的字段。

• async def: 表明这是一个异步函数，可以与 await 一起使用。
  
• self: 这是一个类的方法，因此 self 引用类的实例。
  
• obj_in: DtoType: obj_in 是一个数据传输对象（DTO），它包含了需要插入到数据库中的数据。DtoType 是一个泛型类型，用于表示 DTO 对象。
  
• db: AsyncSession: db 是一个 SQLAlchemy 的异步会话（AsyncSession），用于与数据库进行交互。
  
• **kwargs: 接受恣意数量的关键字参数，允许在对象创建时动态传入额外的字段。
  

• obj_in.model_dump(): 假设 obj_in 是一个 Pydantic 模子或雷同结构，它可以通过 model_dump() 方法转换为字典格式，用于创建 SQLAlchemy 模子实例。
  
• self.model(**obj_in.model_dump(), **kwargs): 使用 obj_in 中的字段以及通过 kwargs 传入的扩展字段来实例化 SQLAlchemy 模子对象。假如 kwargs 非空，它们会被解包并作为额外的字段传入模子构造函数。
  

• db.add(instance): 将新创建的对象添加到当前的数据库会话中。
  
• await db.commit(): 提交事件，将新对象保存到数据库。
  

• SQLAlchemyError: 捕获所有 SQLAlchemy 相关的错误。
  
• await db.rollback(): 在发生非常时，回滚事件，以防止不完整或错误的数据被提交。
  

通过上面的封装，我们可以测试调用的处理例子

1. from crud.customer import customer as customer_crud
2. from models.customer import Customer
3. from pydantic import BaseModel
4. from schemas.customer import CustomerDto, CustomerPageDto
6. async def test_list_customer():
7.     async with get_db() as db:
9.         print("get_list")
10.         totalCount, items = await customer_crud.get_list(
11.             CustomerPageDto(skipCount=0, maxResultCount=10, name="test"),
12.             db,
13.         )
14.         print(totalCount, items)
15.         for customer in customers:
16.             print(customer.name, customer.age)
18.         print("get_by_name")
19.         name = "test"
20.         customer = await customer_crud.get_by_name(
21.             name,
22.             db,
23.         )
24.         if customer:
25.             print(customer.name, customer.age)
26.         else:
27.             print(f"{name} not found")
29.         print("soft delete")
30.         result = await customer_crud.delete_byid(customer.id, db, is_deleted=1)
31.         print("操作结果：", result)
33.         print("soft delete_byids")
34.         result = await customer_crud.delete_byids(
35.             ["11122", "2C5F8672-2AA7-4B14-85AD-DF56F5BF7F1F"], db, is_deleted=1
36.         )
37.         print(f"Soft delete successful: {result}")
39.         print("update_by_column")
40.         result = await customer_crud.update_by_column(
41.             "id", customer.id, {"age": 30}, db
42.         )
43.         print("操作结果：", result)
45.         await db.close()

复制代码

同步和异步处理的差异：

• 同步操作 适用于传统的阻塞式应用场景，比如命令行工具或简单的脚本。
  
• 异步操作 更适合异步框架如 FastAPI，可以提高高并发场景下的性能。
  

通过封装数据库操作，可以让代码更具复用性和可维护性，支持不同类型的操作场景。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4