ORM
ORM：对象关系映射（Object Relational Mapping，简称ORM，或O/RM，或O/R mapping)，用于实现面向对象编程语言里不同类型系统的数据之间的转换
Sqlalchemy
好像是python上运用最多的ORM框架，支持异步同步，多数据库连接，内置缓存机制
官方文档：概述 — SQLAlchemy 1.4 Documentation (osgeo.cn)
结构
db
├── __init__.py
├── curd.py
├── database.py
├── modules.py
└── scheme.py
仅仅作为参考，curd存放具体的增删改查逻辑；database定义数据库类型，位置，同步与异步引擎；modules存放表模型；scheme存放供ORM外的逻辑进行交互的pydantic数据模型，关于为什么要ORM外的数据模型，后面会提到
定义
"""
# database.py
# 同步与异步ORM引擎支持
"""
from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
class DataBase:
"""
- 同步Session
- 异步AsyncSession
"""
SQLALCHEMY_SYNC_DATABASE_URL = "sqlite:///./db.sqlite3"
SQLALCHEMY_ASYNC_DATABASE_URL = "sqlite+aiosqlite:///./db.sqlite3"
engine_sync = create_engine(
SQLALCHEMY_SYNC_DATABASE_URL, connect_args={"check_same_thread": False}
)
engine_async = create_async_engine(
SQLALCHEMY_ASYNC_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionSync = sessionmaker(autocommit=False, autoflush=False, bind=engine_sync)
SessionAsync = sessionmaker(class_=AsyncSession, autocommit=False, autoflush=False, bind=engine_async)
Base = declarative_base()
# 使用sync引擎创建表
@classmethod
def init_database(cls):
# use none async engine to create all table
from .modules import Base
Base.metadata.create_all(bind=cls.engine_sync)
使用pymysql：
mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
表模型
这是我正在使用的一个模型，对应modules.py
class Modules(Base):
""" Saya模组状态 """
__tablename__ = "modules"
id = Column(Integer, primary_key=True, autoincrement=True) # 主键和自增
name = Column(String, index=True, unique=True) # 索引和唯一约束
author = Column(String, nullable=True) # 可为null
description = Column(String, nullable=True)
arg_description = Column(String, nullable=True)
usage = Column(String, nullable=True)
file_name = Column(String, unique=True, index=True)
is_enable = Column(Boolean, default=True) # 默认为true
is_exist = Column(Boolean, default=True)
create_date = Column(Date, default=date.today) # today加（），则默认为启动时的静态时间，不加则为调用时的动态时间
外键表的定义
class BlockListFriend(Base):
""" 模组调用好友黑名单 """
__tablename__ = "block_list_friend"
friend_id = Column(Integer, index=True, primary_key=True)
module = Column(String, ForeignKey("modules.name"), primary_key=True)
ORM外数据模型
注意
你可以不定义对应的pydantic模型，但需要注意增删改查的调用方式，为什么？在注意中会进行说明
from pydantic import BaseModel
from datetime import date, datetime
from typing import Optional
class ModulesBase(BaseModel):
id: Optional[int] = 0
name: str
author: Optional[str] = None
description: Optional[str] = None
arg_description: Optional[str] = None
usage: Optional[str] = None
file_name: str
is_enable: bool = True
is_exist: bool = True
create_date: Optional[date] = None
增删改查
同步方法：
@classmethod
def add_module_sync(cls, db: Session, file_name: str, ....):
new_module = Moudles(file_name=file_name, .....)
db.add(new_module)
db.commit()
@classmethod
def get_module_by_file_name_sync(cls, db: Session, file_name: str) -> Optional[Modules]:
module = db.query(Modules).filter_by(file_name=file_name).one_or_none()
return module
@classmethod
def update_module_sync(cls, db: Session, file_name: str, somevalue:....):
db.query(Modules).filter_by(file_name=file_name).update({"key":"values"})
db.commit()
@classmethod
def delete_module_sync(cls, db: Session, file_name: str):
db.query(Modules).filter_by(file_name=file_name).delete()
db.commit()
异步方法：
@classmethod
async def get_module_by_name_async(cls, db: AsyncSession, module: str) -> Optional[ModulesBase]:
res = await db.execute(select(Modules).where(Modules.name == module).limit(1))
module = res.one_or_none()
if module:
return ModulesBase(**module[0].__dict__)
return None
@classmethod
async def insert_module_async(cls, db: AsyncSession, name: str, author: str, description: str,
arg_description: str, usage: str, file_name: str, init_enable_status: bool = True):
await db.execute(insert(Modules).values(
name=name,
author=author,
description=description,
arg_description=arg_description,
usage=usage,
file_name=file_name))
await db.commit()
@classmethod
async def update_module_async(cls, db: AsyncSession, module_dict: dict):
update_module = update(Modules).where(Modules.id == id_).values(module_dict)
await db.execute(update_module)
await db.commit()
@classmethod
async def delete_module_by_name(cls, db: AsyncSession, moudle_name: str):
await db.execute(delete(Modules).where(Modules.name == moudle_name))
await db.commit()
操作上需要使用
async with DataBase.SessionAsync() as db:
以获取异步session对象，同步则是
with DataBase.SessionSync() as db:
sort、count、join等操作
先放一个我已经不打算推进的项目方法：
_order_dict = {
1: models.Work.release,
2: models.Work.add_date,
3: models.Work.dl_count,
4: models.Work.price,
5: models.Work.review_count,
6: models.Work.rate_average,
7: models.work_review.c.rating
}
@classmethod
async def get_works(cls, db: Session, user_id: int, order: int, nsfw: bool = True, page: int = 1, sort: int = 1):
"""
get all works、按一定规则得到所有作品
Order(排序规则)：
1: by release time,按照发售日期
2: by add date,按照收录入库日期
3: by dl count,按照售出数量
4: by price, 按照价格
5: by review count,按照评论数量
6: by rate average,按照平均评分
7: by user rate,按照用户对作品的评分
Sort(升降):
0: Asce,升序
1: Desc,降序
"""
time1 = process_time()
# order = text("-" + cls._order_dict[order] if sort else cls._order_dict[order])
order = desc(cls._order_dict[order]) if sort else cls._order_dict[order]
execute = select(
models.Work.id,
models.Work.title,
models.Circle,
models.Work.nsfw,
models.Work.release,
models.Work.dl_count,
models.Work.price,
models.Work.review_count,
models.Work.rate_count,
models.Work.rate_average,
models.Work.rate_count_detail,
models.work_review.c.review_text.label('review'),
models.work_review.c.rating.label('rating'),
models.work_review.c.created_date.label('review_date'),
models.work_review.c.progress.label('progress'),
).outerjoin(
models.Circle,
models.Work.circle_id == models.Circle.id,
).outerjoin(
models.work_review,
models.work_review.c.user_id == user_id,
)
if not nsfw:
execute = execute.where(models.Work.nsfw == False)
execute = execute.order_by(order).slice(
(page - 1) * settings.config.page_size, page * settings.config.page_size
)
res = await db.execute(execute)
works = res.all()
out_works = []
work_lens = await db.execute(select(func.count(models.Work.id)))
work_lens = work_lens.first()[0]
for work in works:
tags = await db.execute(select(models.Tag).join(
models.work_tag
).where(
models.work_tag.c.work_id == work.id,
models.Tag.id == models.work_tag.c.tag_id
))
tags = [tag[0] for tag in tags.all()]
s = schemas.Work(
**work,
tags=tags,
)
out_works.append(s)
print(f"process_time: {process_time() - time1}")
return schemas.Works(
work_list=out_works,
pagination=schemas.Pagination(
current_page=page,
page_size=settings.config.page_size,
totalCount=work_lens
)
)
select(func.count(models.Work.id)计数await db.execute(select(models.Tag).join(models.work_tag)多表order_by()desc(module.key)排序module.key.label('name')对某个字段重命名slice((page - 1) * settings.config.page_size, page * settings.config.page_size)使用偏移进行分页
可见实现一个复杂的查询还是不怎么轻松的，代码量也不会小，不过，这个方法本身是很不堪入目的，所有操作都挤在一起，没有对逻辑进行切割，这里仅仅作为展示和记录
注意
异步操作方法返回的不是模组对象本身
注意到await后面还加了个[0]么，这是因为他们返回的都是tuple类型，也就是：
return type : (module, )
查询多值的时候：
return type : [(module, ), (module, ), (module, ) ....]
业务数据与ORM数据分离
为什么呢，先给个例子：
async def all_test():
db = DataBase.SessionAsync()
res = await db.execute(select(modules.Modules))
module = res.all()
n = 0
for i in module:
print(f"数据：{module[n][0].__dict__}")
await db.execute(update(modules.Modules).where(modules.Modules.id==i[0].id).values({"name": f"t1t11{n}"}))
await db.commit()
n += 1
await db.close()
loop = asyncio.get_event_loop()
loop.run_until_complete(all_test())
拿到列表后遍历改变name字段，很普通的逻辑，看起来挺正常的对吧，但是一运行就会报错：
File "C:\Users\name\AppData\Local\pypoetry\Cache\virtualenvs\project-VkH3dFMM-py3.9\lib\site-packages\sqlalchemy\util\_concurrency_py3k.py", line 67, in await_only
raise exc.MissingGreenlet(
sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called; can't call await_() here. Was IO attempted in an unexpected place? (Background on this error at: https://sqlalche.me/e/14/xd2s)
查看下print()出来的内容：
数据：{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x048BB748>, 'id': 2, 'usage': None, ...}
数据：{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x048BB778>}
是不是少了什么，第一次遍历的模型实例数据清晰可见，但第二个模型实例的数据呢？它除了一个InstanceState对象之外数据全丢失了！
由此可以得出结论：
当commit操作结束后，所有得到的模型实例都会被刷新掉，只剩下一个没有数据的空壳对象，这就是为什么建议对业务数据和ORM数据分离，把模型实例的数据用pydantic模型等保存下来才不会被commit操作刷新掉数据
多对多关系的第三表
work_tag = Table(
'work_tag',
Base.metadata,
Column("work_id", Integer, ForeignKey("works.id"), primary_key=True),
Column("tag_id", Integer, ForeignKey("tags.id"), primary_key=True),
)
class Tag(Base):
__tablename__ = "tags"
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String, unique=True)
# works = relationship("Work", secondary=work_tag, back_populates="tags", lazy="dynamic")
works = relationship("Work", secondary=work_tag, lazy="dynamic")
class Work(Base):
__tablename__ = "works"
id = Column(Integer, primary_key=True)
root_folder = Column(String)
dir = Column(String)
title = Column(String)
....
tags = relationship("Tag", secondary=work_tag, back_populates="works", lazy="dynamic")
这是定义多对多第三表的方式，当然你若是不用relationship去定义关系映射，而只使用ForeignKey的话，是可以不使用这种方式的