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

from sqlalchemy importColumn, Integer, String, ForeignKeyfrom sqlalchemy.orm importrelationship, declarative_basefrom sqlalchemy.ext.asyncio importcreate_async_engine, AsyncSessionfrom sqlalchemy.orm importsessionmaker



Base=declarative_base()classDictTypeInfo(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)



asyncdefinit_db():

    async with engine.begin() as conn:

        await conn.run_sync(Base.metadata.create_all)#示例：如何插入数据并进行查询
async defexample_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 inparent_nodes:print(f"Parent Node: {node.name}, Children: {[child.name for child in node.children]}")

async defget_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 defexample_usage():

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

from sqlalchemy importTable, Column, Integer, ForeignKeyfrom sqlalchemy.ext.declarative importdeclarative_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 importrelationshipclassLeftModel(Base):__tablename__ = 'left_table'id= Column(Integer, primary_key=True)

    name= Column(String(50))

    rights= relationship("RightModel", secondary=association_table, back_populates="lefts")classRightModel(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 importcreate_enginefrom sqlalchemy.orm importsessionmaker#创建数据库引擎
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 inleft1.rights:print(right.name)  #输出: Right 1, Right 2

for left inright1.lefts:print(left.name)  #输出: Left 1

#添加关系
left1.rights.append(RightModel(name="Right 3"))

session.commit()#删除关系
left1.rights.remove(right2)

session.commit()

from sqlalchemy.future importselectfrom sqlalchemy.orm importaliasedfrom sqlalchemy.ext.asyncio importAsyncSessionfrom sqlalchemy.ext.asyncio importcreate_async_enginefrom sqlalchemy.orm importsessionmaker#假设你的数据库模型是 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)



asyncdefget_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()returndict_data#示例用法
importasyncio



asyncdefexample_usage():

    dict_type_name= "some_type_name"dict_data=await get_dict_data(dict_type_name)for data indict_data:print(data)

#定义 CListItem 类
classCListItem:def __init__(self, name, value):

        self.name=name

        self.value=value#定义示例列表和转换操作
defconvert_list_items(list_items):

    dict_list=[]if list_items:  #确保 list_items 不是 None
        for info inlist_items.Items:

            dict_list.append(CListItem(info.Name, info.Value))return dict_list

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

from sqlalchemy importColumn, Stringfrom sqlalchemy.ext.declarative importdeclarative_base



Base=declarative_base()classMyModel(Base):__tablename__ = 'my_table'id= Column(String(36), primary_key=True, default=generate_sequential_uuid, nullable=False)#其他字段...

from sqlalchemy importColumn, Integer, Stringfrom sqlalchemy.ext.declarative importdeclarative_base



Base=declarative_base()classMyModel(Base):__tablename__ = 'my_table'id= Column(Integer, primary_key=True, autoincrement=True)

    name= Column(String(50))#其他字段..

from sqlalchemy importSequenceclassMyModel(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() ifline]#初始化变量
int_seq = -1seq_length= 3str_seq=input_seq#尝试将 str_seq 转换为整数
ifstr_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}")

importre#假设 info 是一个包含 Name 和 Value 属性的对象
classInfo: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 elsearray[0]#打印结果
print(f"Name: {info.Name}")print(f"Value: {info.Value}")

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

importre



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