I'm new to SQLAlchemy, and ran into an issue when I tried to create __init__ function inside the table class. The script can run without exception but the association function won't take any values after I add the object to database (as an indicator the output results have "0" length of related attributes). However, as soon as I remove the __init__ function everything works as desired.
I wonder what the cause of this issue is, and if there is a work around that allows me to have some hints/arguments validator when insert new data.
working script:
Base = declarative_base()
a_b = Table('a_b', Base.metadata,
Column('a_id', ForeignKey('a.a_id')),
Column('b_id', ForeignKey('b.b_id'))
)
class A(Base):
__tablename__ = 'a'
a_id = Column(Integer, primary_key=True)
a_age = Column(Integer)
b_lst = relationship('B', secondary=a_b, backref='a_lst')
def __repr__(self):
return f'A({self.a_age} >> length of b_lst: {len(self.b_lst)})'
class B(Base):
__tablename__ = 'b'
b_id = Column(Integer, primary_key=True)
b_num = Column(Integer)
def __repr__(self):
return f'B({self.b_num} >> length of a_lst: {len(self.a_lst)})'
database_path = 'sqlite:///../data/database.db'
engine = create_engine(url=database_path, echo=False, future=True)
Base.metadata.create_all(engine)
with Session(engine) as session:
a_lst = [
A(a_age=1, b_lst=[B(b_num=11), B(b_num=22)]),
A(a_age=2, b_lst=[B(b_num=33)])
]
session.add_all(a_lst)
session.commit()
stmt = select(A)
results = session.scalars(stmt)
pprint([item for item in results])
- results
[A(1 >> length of b_lst: 2), A(2 >> length of b_lst: 1)]
issue script:
Base = declarative_base()
a_b = Table('a_b', Base.metadata,
Column('a_id', ForeignKey('a.a_id')),
Column('b_id', ForeignKey('b.b_id'))
)
class A(Base):
__tablename__ = 'a'
a_id = Column(Integer, primary_key=True)
a_age = Column(Integer)
b_lst = relationship('B', secondary=a_b, backref='a_lst')
def __init__(self, a_age, *args, **kwargs):
self.a_age = a_age
def __repr__(self):
return f'A({self.a_age} >> length of b_lst: {len(self.b_lst)})'
class B(Base):
__tablename__ = 'b'
b_id = Column(Integer, primary_key=True)
b_num = Column(Integer)
def __init__(self, b_num, *args, **kwargs):
self.b_num = b_num
def __repr__(self):
return f'B({self.b_num} >> length of a_lst: {len(self.a_lst)})'
database_path = 'sqlite:///../data/database.db'
engine = create_engine(url=database_path, echo=False, future=True)
Base.metadata.create_all(engine)
with Session(engine) as session:
a_lst = [
A(a_age=1, b_lst=[B(b_num=11), B(b_num=22)]),
A(a_age=2, b_lst=[B(b_num=33)])
]
session.add_all(a_lst)
session.commit()
stmt = select(A)
results = session.scalars(stmt)
pprint([item for item in results])
- results
[A(1 >> length of b_lst: 0), A(2 >> length of b_lst: 0)]