在 SQLAlchemy 中实现数据处理的时候，实现表自引用、多对多、联合查询，有序id等常见的一些经验总结

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, ForeignKey('dict_type_info.id'))  # 外键指向父节点的 id

    # 定义 parent 关系
    parent = relationship("DictTypeInfo", remote_side=[id], back_populates="children")

    # 定义 children 关系
    children = relationship("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, parent=parent_node)
            child_node2 = DictTypeInfo(name="Child2", code="C002", remark="Second Child", seq=2, parent=parent_node)
            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}, Children: {[child.name for child in node.children]}")

async def get_tree(pid: int):
    async with AsyncSessionLocal() as session:
        # 通过 selectinload 加载所有子节点
        stmt = select(DictTypeInfo).filter(DictTypeInfo.pid == pid).options(selectinload(DictTypeInfo.children))
        result = await session.execute(stmt)
        nodes = result.scalars().all()
        
        return nodes

async def example_usage():
    nodes = await get_tree(pid=1)
    for node in nodes:
        print(f"Node: {node.name}, Children: {[child.name for child in node.children]}")

from sqlalchemy import Table, Column, Integer, ForeignKey
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

association_table = Table('association', Base.metadata,
    Column('left_id', Integer, ForeignKey('left_table.id')),
    Column('right_id', Integer, ForeignKey('right_table.id'))
)

from sqlalchemy.orm import relationship

class LeftModel(Base):
    __tablename__ = 'left_table'

    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    rights = relationship("RightModel", secondary=association_table, back_populates="lefts")

class RightModel(Base):
    __tablename__ = 'right_table'

    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    lefts = relationship("LeftModel", secondary=association_table, back_populates="rights")

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 = [right1, right2]

# 添加到会话并提交
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()

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 get_dict_data(dict_type_name: str):
    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 get_dict_data(dict_type_name)
    for data in dict_data:
        print(data)

# 定义 CListItem 类
class CListItem:
    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:
            dict_list.append(CListItem(info.Name, info.Value))
    return dict_list

def generate_sequential_uuid():
    return uuid.uuid1()  # 基于时间生成有序UUID

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)
    # 其他字段...

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))
    # 其他字段..

from sqlalchemy import Sequence

class MyModel(Base):
    __tablename__ = 'my_table'

    id = Column(Integer, Sequence('my_sequence'), primary_key=True)
    name = Column(String(50))

# 假设 Data 和 input.Seq 是从输入中获取的
Data = "example\nline1\nline2\n"  # 示例数据
input_seq = "123"  # 示例序列字符串

# 将 Data 按行拆分，并移除空行
array_items = [line for line in Data.splitlines() if line]

# 初始化变量
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}")

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[0]
info.Value = array[1] if len(array) > 1 else array[0]

# 打印结果
print(f"Name: {info.Name}")
print(f"Value: {info.Value}")

str_item = " 1,2,3;4;5/6/7、8、9；10 "

import re

result = re.split(r"[;,|/，；、]+", str_item.strip())
print(result)