在 SQLAlchemy 中实现数据处理的时间，实现表自引用、多对多、联合查询，有
有时间，我们在利用SQLAlchemy操纵某些表的时间，需要利用外键关系来实现一对多或者多对多的关系引用，以及对多表的联合查询，有序列的uuid值或者自增id值，字符串的分拆等常见处理操纵。1、在 SQLAlchemy 中界说具有嵌套 children 关系的表
要在 SQLAlchemy 中界说具有嵌套 children 关系的表，如表中包含 id 和 pid 字段，可以利用 relationship 和 ForeignKey 来建立父子关系。
首先，你需要界说一个模子类，此中包含 id 和 pid 字段。id 是主键，pid 是指向父记载的外键。然后，你利用 relationship 来建立父子关系。
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, declarative_base
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
class DictTypeInfo(Base):
__tablename__ = 'dict_type_info'
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
code = Column(String)
remark = Column(String)
seq = Column(Integer)
pid = Column(Integer, <strong>ForeignKey('dict_type_info.id')</strong>)# 外键指向父节点的 id
# 定义 parent 关系
<strong>parent</strong> = <strong>relationship</strong>("DictTypeInfo", remote_side=, back_populates="children")
# 定义 children 关系
<strong>children</strong> = <strong>relationship</strong>("DictTypeInfo", back_populates="parent")例子利用代码如下所示。
# 创建异步引擎和会话
DATABASE_URL = "mysql+asyncmy://username:password@localhost/mydatabase"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)
async def init_db():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
# 示例：如何插入数据并进行查询
async def example_usage():
async with AsyncSessionLocal() as session:
async with session.begin():
# 插入数据
parent_node = DictTypeInfo(name="Parent", code="P001", remark="Parent Node", seq=1)
child_node1 = DictTypeInfo(name="Child1", code="C001", remark="First Child", seq=1, <strong>parent=</strong><strong>parent_node</strong>)
child_node2 = DictTypeInfo(name="Child2", code="C002", remark="Second Child", seq=2, <strong>parent=</strong><strong>parent_node</strong>)
session.add(parent_node)
session.add(child_node1)
session.add(child_node2)
# 查询数据
async with session.begin():
result = await session.execute(
"SELECT * FROM dict_type_info WHERE pid IS NULL"
)
parent_nodes = result.scalars().all()
for node in parent_nodes:
print(f"Parent Node: {node.name}, <strong>Children</strong>: {<strong></strong>}")代码说明
[*]界说模子类 (DictTypeInfo):
[*]id: 主键。
[*]pid: 外键，指向同一个表的 id，表示父节点。
[*]parent: 父关系，通过 remote_side 设定本模子的外键指向自身的主键。
[*]children: 子关系，back_populates 用于双向关系的映射。
[*]创建异步引擎和会话:
[*]利用 create_async_engine 和 AsyncSession 创建数据库引擎和会话，以支持异步操纵。
[*]插入和查询数据:
[*]插入数据示例展示了怎样创建父节点和子节点，并将子节点关联到父节点。
[*]查询数据示例展示了怎样查询所有父节点以及它们的子节点。
注意事项
[*]remote_side: 在 relationship 中，remote_side 是指定哪些字段是远程的一方（即子节点关系的目标）。
[*]确保在模子中界说了正确的外键约束。在你提供的模子中，pid 列需要指向同一表中的 id 列。确保 ForeignKey 设置正确。
[*]异步操纵: 利用 AsyncSession 和 asyncio 举行异步数据库操纵。
[*]创建表: 在初始化数据库时，确保表结构是正确的。
要利用 selectinload 加载某个 pid 下的对象及其子列表，可以通过 SQLAlchemy 的 selectinload 来优化加载子关系。selectinload 可以减少 SQL 查询的数量，特别是在加载具有层次结构的数据时。
async def get_tree(pid: int):
async with AsyncSessionLocal() as session:
# 通过 selectinload 加载所有子节点
stmt = select(DictTypeInfo).filter(DictTypeInfo.pid == pid).options(<strong>selectinload(DictTypeInfo.children)</strong>)
result = await session.execute(stmt)
nodes = result.scalars().all()
return nodes如许，调用 get_tree 函数获取指定 pid 的节点及其子节点，代码如下。
async def example_usage():
nodes = <strong>a<strong>w</strong>ait get_tree(pid=1)
</strong>for node in nodes:
print(f"Node: {node.name}, Children: {}")selectinload: selectinload 可以减少 N+1 查询问题，它通过一条额外的查询来加载相关对象。这适合用于层次结构数据的加载。通过这种方式，你可以利用 SQLAlchemy 的 selectinload 来高效地加载具有父子关系的对象，并优化数据库查询性能。
同样，我们在 SQLAlchemy 中实现多对多关系也是雷同的处理方式。
在 SQLAlchemy 中，实现多对多关系通常需要创建一个关联表（association table），该表将存储两个相关联表的外键，从而实现多对多关系。以下是一个实现多对多关系的详细步骤。
1） 界说多对多关系的关联表
首先，需要界说一个关联表，该表包含两个外键，分别指向两头的主表。这通常利用 Table 对象来实现。
from sqlalchemy import Table, Column, Integer, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
<strong>association_table</strong> = Table('association', Base.metadata,
Column('left_id', Integer, ForeignKey('left_table.id')),
Column('right_id', Integer, ForeignKey('right_table.id'))
)在这个例子中，association_table 是一个包含两个外键的中间表：left_id 和 right_id 分别指向 left_table 和 right_table 的主键。
2）界说两头的模子并添加关系
在两头的模子中，利用 relationship 来界说多对多关系，并指定 secondary 参数为关联表。
from sqlalchemy.orm import relationship
class<strong> LeftModel</strong>(Base):
__tablename__ = 'left_table'
id = Column(Integer, primary_key=True)
name = Column(String(50))
rights = <strong>relationship</strong>("RightModel", secondary=association_table, back_populates="lefts")
class<strong> RightModel</strong>(Base):
__tablename__ = 'right_table'
id = Column(Integer, primary_key=True)
name = Column(String(50))
lefts = <strong>relationship</strong>("LeftModel", secondary=association_table, back_populates="rights")
[*]rights 是 LeftModel 中界说的关系属性，它将毗连到 RightModel。
[*]lefts 是 RightModel 中界说的关系属性，它将毗连到 LeftModel。
[*]secondary=association_table 告诉 SQLAlchemy 利用 association_table 作为毗连表。
[*]back_populates 用于双向关系的对称引用。
3）创建数据库并插入数据
下面的代码展示了怎样创建数据库、插入数据并查询多对多关系。
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# 创建数据库引擎
engine = create_engine('sqlite:///example.db')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
# 创建模型实例
left1 = LeftModel(name="Left 1")
right1 = RightModel(name="Right 1")
right2 = RightModel(name="Right 2")
# 设置多对多关系
left1.rights =
# 添加到会话并提交
session.add(left1)
session.commit()
# 查询并打印关系
for right in left1.rights:
print(right.name)# 输出: Right 1, Right 2
for left in right1.lefts:
print(left.name)# 输出: Left 1你可以像操纵普通列表一样来处理这些关系，例如添加、删除关联等：
# 添加关系
left1.rights.append(RightModel(name="Right 3"))
session.commit()
# 删除关系
left1.rights.remove(right2)
session.commit() 通过这些步骤，你可以在 SQLAlchemy 中实现和操纵多对多关系。
2、在 SQLAlchemy 中联合多个表举行记载关联查询
例如，在我的框架中，字典大类和字典项目是差异的表进管理的，因此如果需要根据大类名称举行字典项目的查询，那么就需要联合两个表举行处理。
详细操纵如下：创建一个查询，将 DictDataInfo 表与 DictTypeInfo 表联接（通过 DictType_ID 和 Id 列）
from sqlalchemy.future import select
from sqlalchemy.orm import aliased
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker
# 假设你的数据库模型是 DictDataInfo 和 DictTypeInfo
# 需要提前定义好这两个模型类
DATABASE_URL = "mysql+asyncmy://username:password@localhost/mydatabase"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)
async def<strong> get_dict_data(dict_type_name: str):</strong>
async with AsyncSessionLocal() as session:
# 创建别名
DictData = aliased(DictDataInfo)
DictType = aliased(DictTypeInfo)
# 联合查询并根据条件过滤
stmt = (
select(DictData)
.join(DictType, DictData.DictType_ID == DictType.id)
.filter(DictType.name == dict_type_name)
)
result = await session.execute(stmt)
dict_data = result.scalars().all()
return dict_data
# 示例用法
import asyncio
async def example_usage():
dict_type_name = "some_type_name"
dict_data = await <strong>get_dict_data</strong>(dict_type_name)
for data in dict_data:
print(data)代码说明
[*]aliased: 利用 aliased 创建表的别名，如许可以方便地在查询中引用这些表。
[*]join: 利用 join 举行表毗连。这里 DictDataInfo 表的 DictType_ID 列与 DictTypeInfo 表的 id 列毗连。
[*]filter: 利用 filter 来添加条件筛选，筛选出 DictTypeInfo 表中 name 列等于 dict_type_name 的记载。
[*]select: 利用 select 语句来选择 DictDataInfo 表中的记载，这对应于 Select(d => d)。
[*]异步操纵: 由于利用的是 SQLAlchemy 的异步模式，所有数据库操纵都在 async with 和 await 语句中举行，以确保异步执行。
如果我们需要将得到的数据举行对象转换，我们可以利用下面的处理代码实现。
# 定义 CListItem 类
class<strong> CListItem</strong>:
def __init__(self, name, value):
self.name = name
self.value = value
# 定义示例列表和转换操作
def convert_list_items(list_items):
dict_list = []
if list_items:# 确保 list_items 不是 None
for info in list_items.Items:
<strong>dict_list.append</strong>(CListItem(info.Name, info.Value))
return dict_list
3、利用sqlalchemy插入数据的时间，怎样判断为非自增类型的时间，id赋值一个有序列的uuid值
有时间，我们的数据表主键是用字符串的，这种适用于很广的用途，比较容易在插入的时间就确定好id键的值，从而可以处理相关的内容。
但是，有时间我们可以让后端举行确定一个有序的ID值，那么利用SQLAlchemy 我们应该怎样实现？
首先，确保你已经导入了 uuid 库，这是用于生成 UUID 的 Python 标准库。
有序 UUID 通常是基于时间的 UUID。你可以利用 uuid.uuid1() 来生成基于时间的 UUID。
def generate_sequential_uuid():
return uuid.uuid1()# 基于时间生成有序UUID在界说 SQLAlchemy 模子时，可以将 id 字段设置为利用该函数生成的 UUID。通常在模子中通过 default 参数设置默认值。
from sqlalchemy import Column, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class MyModel(Base):
__tablename__ = 'my_table'
id = Column(String(36), primary_key=True, default=generate_sequential_uuid, nullable=False)
# 其他字段...在插入新数据时，如果 id 字段为空，它将自动利用 generate_sequential_uuid 函数生成一个基于时间的 UUID。
如许就可以确保在插入数据时，非自增类型的 id 字段会被赋值为一个有序列的 UUID 值。
对于自增的整型 id，SQLAlchemy 提供了自动处理机制。你只需要在模子中将 id 字段界说为 Integer 类型，并设置 primary_key=True，SQLAlchemy 就会自动为该字段设置自增属性。
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class MyModel(Base):
__tablename__ = 'my_table'
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(50))
# 其他字段..默认情况下，SQLAlchemy 会利用数据库的原生自增机制（如 MySQL 的 AUTO_INCREMENT 或 PostgreSQL 的 SERIAL）。如果你需要利用自界说的自增策略，可以通过设置 Sequence 来实现（适用于支持 Sequence 的数据库，如 PostgreSQL）。
from sqlalchemy import Sequence
class MyModel(Base):
__tablename__ = 'my_table'
id = Column(Integer, Sequence('my_sequence'), primary_key=True)
name = Column(String(50))在上述代码中，Sequence('my_sequence') 界说了一个序列，SQLAlchemy 将利用该序列生成自增的 id 值。
通过这些步骤，你可以轻松处理整型自增 id 字段，SQLAlchemy 会自动为每个新记载分配唯一的自增 id。
4、在插入记载的时间，对字符串的数据处理
在批量插入数据字典的时间，我希望根据用户输入内容（多行数据）举行转化，把每行的数据分拆举行判断，如果符合条件的举行处理插入。
在 Python 中，可以利用字符串的 splitlines() 方法来实现相同的功能。
# 假设 Data 和 input.Seq 是从输入中获取的
Data = "example\nline1\nline2\n"# 示例数据
input_seq = "123"# 示例序列字符串
# 将 Data 按行拆分，并移除空行
array_items =
# 初始化变量
int_seq = -1
seq_length = 3
str_seq = input_seq
# 尝试将 str_seq 转换为整数
if str_seq.isdigit():
int_seq = int(str_seq)
seq_length = len(str_seq)
# 打印结果
print(f"Array Items: {array_items}")
print(f"int_seq: {int_seq}")
print(f"seq_length: {seq_length}")
[*]Python 的 splitlines() 方法将字符串按行分割，同时自动处理各种换行符（包括 \n 和 \r\n）。
[*]列表推导式 移除了空行，雷同于 C# 中的 StringSplitOptions.RemoveEmptyEntries。
[*]利用 str_seq.isdigit() 检查 str_seq 是否全部由数字组成，这雷同于 C# 的 int.TryParse。
在 Python 中，可以利用 re.split() 函数来按照正则表达式分割字符串。以下是对应的 Python 代码：
import re
# 假设 info 是一个包含 Name 和 Value 属性的对象
class Info:
def __init__(self):
self.Name = ""
self.Value = ""
info = Info()
# dictData 是输入的字符串
dict_data = "example_name example_value"
# 使用正则表达式按照空白字符分割字符串
array = re.split(r'\s+', dict_data)
# 赋值给 info 对象的属性
info.Name = array
info.Value = array if len(array) > 1 else array
# 打印结果
print(f"Name: {info.Name}")
print(f"Value: {info.Value}")利用 re.split() 函数根据空白字符（包括空格、制表符等）分割字符串 dict_data。r'\s+' 是一个正则表达式，表示一个或多个空白字符。
如果你需要根据多个分隔符来分割字符串，同样可以利用正则表达式（re 模块）的 re.split() 方法。
str_item = " 1,2,3;4;5/6/7、8、9；10 "
import re
result = re.split(r"[;,|/，；、]+", str_item.strip())
print(result)结果输出：['1', '2', '3', '4', '5', '6', '7', '8', '9', '10']
解释：
[*]re.split(r'[;,|/，；、]', text) 中的 r'[;,|/，；、]' 是一个正则表达式模式：
[] 表示字符类，表示匹配字符类中的任意一个字符。
;,|/，；、 分别表示分号、逗号，竖线，中文逗号，中文分号，和空格，这些字符都将作为分隔符。
利用正则表达式可以灵活处理多个分隔符，适用于更复杂的分割需求。
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]