# encoding=utf-8
print(rf"{'程序开始':*^120}")
from random import choice, randint
from typing import Optional, Union
from sqlalchemy import Table, create_engine, text,ForeignKey,Column,String,Text,DECIMAL,Enum,Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker,relationship,backref
url=rf"mysql+pymysql://root:root@localhost:3306"
enginesqlalchemy=create_engine(url=url)
connection_sqlalchemy=enginesqlalchemy.connect()
databasename="sgz"
connection_sqlalchemy.execute(statement=text(text=rf"create database if not exists {databasename} default character set utf8;"))
connection_sqlalchemy.close()
engine_sqlalchemy=create_engine(url=rf'{url}/{databasename}')
sessionClassSqlalchemy=sessionmaker(bind=engine_sqlalchemy)
session_sqlalchemy=sessionClassSqlalchemy()
BaseclassSqlalchemy=declarative_base()
# 建立操作Table对象需要用的连接(session_sqlalchemy只适用于orm模型,而Table对象不是orm模型)
connecttionsqlalchemy=engine_sqlalchemy.connect()
# region
# class MiddleTable(BaseclassSqlalchemy):
# """_summary_
# Args:
# BaseclassSqlalchemy (_type_): _description_
# """
# __tablename__="中间表"
# ip=Column(Integer,primary_key=True,autoincrement=True,default=0,name="序号")
# uip=Column(Integer,ForeignKey(column=rf"主表.序号"),primary_key=True)
# uuip=Column(Integer,ForeignKey(column=rf"副表.序号"),primary_key=True)
# def __repr__(self):
# """_summary_
# Returns:
# _type_: _description_
# """
# return rf"<ip:{self.ip} uip:{self.uip} uuip:{self.uuip}>"
# endregion
MiddleTable=Table("多对多中间表",BaseclassSqlalchemy.metadata,
Column(Integer,primary_key=True,autoincrement=True,name="序号"),
Column(Integer,ForeignKey(column=f"主表.序号",ondelete="CASCADE"),primary_key=True,name="主表外键"),
Column(Integer,ForeignKey(column=f"副表.序号",ondelete="CASCADE"),primary_key=True,name="副表外键"))
class MainTable(BaseclassSqlalchemy):
"""_summary_
Args:
BaseclassSqlalchemy (_type_): _description_
"""
__tablename__="主表"
ip=Column(Integer,primary_key=True,autoincrement=True,name="序号")
name=Column(String(length=255),name="姓名")
introduction=Column(Text(length=65535),name="生平")
region=Column(Enum("汉","魏","蜀","吴","晋","乌桓","鲜卑","匈奴","羌","氐",
"羯","山越","南蛮"),name="地区")
subordianterelationship=relationship(argument="SubordianteTable",
secondary=MiddleTable,
backref=backref(name="mainbackreference",uselist=False),uselist=False)
def __repr__(self):
"""_summary_
Returns:
_type_: _description_
"""
return rf"<ip:{self.ip} name:{self.name} introduction:{self.introduction} region:{self.region}>"
class SubordianteTable(BaseclassSqlalchemy):
"""_summary_
Args:
BaseclassSqlalchemy (_type_): _description_
"""
__tablename__="副表"
ip=Column(Integer,primary_key=True,autoincrement=True,name="序号")
price=Column(DECIMAL(precision=12,scale=2),name="价值",default=0)
def __repr__(self):
"""_summary_
Returns:
_type_: _description_
"""
return rf"<ip:{self.ip} price:{self.price}>"
BaseclassSqlalchemy().metadata.create_all(bind=engine_sqlalchemy)
maininstances=[]
subordinateinstances=[]
middleinstances=[]
for i in range(1,20):
maindata=MainTable(name=rf"姓名{i}",introduction=rf"介绍{i}",
region=choice(seq=("汉","魏","蜀","吴","晋","乌桓",
"鲜卑","匈奴","羌","氐","羯","山越","南蛮")))
maininstances.append(maindata)
subordinatedata=SubordianteTable(price=randint(a=i,b=i*100000))
subordinateinstances.append(subordinatedata)
# print(f"{'maininstances':★^120}\n{maininstances}")
# print(f"{'subordinateinstances':★^120}\n{subordinateinstances}")
# print(f"{'middleinstances':★^120}\n{middleinstances}")
def add_data_by_Table():
"""_summary_
Returns:
_type_: _description_
"""
session_sqlalchemy.add_all(instances=maininstances)
session_sqlalchemy.add_all(instances=subordinateinstances)
session_sqlalchemy.commit()
for maindata,subordinatedata in zip(maininstances,subordinateinstances):
middledata = MiddleTable.insert().values(主表外键=maindata.ip, 副表外键=subordinatedata.ip)
connecttionsqlalchemy.execute(middledata)
connecttionsqlalchemy.commit()
return None
def query_data_by_table(filtercondition_parameter:Optional[Union[str,int]])->any:
"""_summary_
Args:
filtercondition_parameter (Optional[str,int]): _description_
Returns:
any: _description_
"""
results_maintable=session_sqlalchemy.query(MainTable).filter(MainTable.ip>filtercondition_parameter).all()
print(f"{'副表结果--来自主表查询':*^120}")
# print(results_maintable)
for result in results_maintable:
print(result.subordianterelationship)
results_subordiantetable=session_sqlalchemy.query(SubordianteTable).filter(SubordianteTable.ip>filtercondition_parameter).all()
print(f"{'主表结果--来自副表查询':*^120}")
# print(results_subordiantetable)
for result in results_subordiantetable:
print(result.mainbackreference)
return results_maintable,results_subordiantetable
add_data_by_Table()
query_data_by_table(filtercondition_parameter=0)
session_sqlalchemy.close()
connection_sqlalchemy.close()
connecttionsqlalchemy.close()
if __name__=="__main__":
print(rf"{'程序结束':*^120}")
老师好,以上是我的代码,运行后遇到一个奇怪的bug,GPT-4也解决不了:代码不会报错,能正常运行,但是第一次运行后无法显示任何结果,第二次运行只能显示第一次操作时添加的数据,第三次运行只能显示上一轮操作后的结果
您能帮忙看下问题吗