Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import sqlalchemy as sa
- from sqlalchemy.ext.declarative import declarative_base
- from sqlalchemy.orm import relationship, sessionmaker
- engine = sa.create_engine(
- r"mssql+pyodbc://@localhost:49242/myDb?driver=ODBC+Driver+17+for+SQL+Server"
- )
- with engine.connect() as conn:
- for tb_name in ["tbl_a", "tbl_c", "tbl_b"]:
- conn.execute(sa.text(f"DROP TABLE IF EXISTS [{tb_name}]"))
- Base = declarative_base()
- class A(Base):
- __tablename__ = "tbl_a"
- id = sa.Column(sa.Integer(), primary_key=True)
- b_id = sa.Column(sa.Integer(), sa.ForeignKey("tbl_b.id"))
- c_id = sa.Column(sa.Integer(), sa.ForeignKey("tbl_c.id"))
- related_b = relationship("B", back_populates="related_as")
- related_c = relationship("C", back_populates="related_as")
- class B(Base):
- __tablename__ = "tbl_b"
- id = sa.Column(sa.Integer(), primary_key=True)
- related_cs = relationship("C", back_populates="related_b")
- related_as = relationship("A", back_populates="related_b")
- class C(Base):
- __tablename__ = "tbl_c"
- id = sa.Column(sa.Integer(), primary_key=True)
- b_id = sa.Column(sa.Integer(), sa.ForeignKey("tbl_b.id"))
- related_b = relationship("B", back_populates="related_cs")
- related_as = relationship("A", back_populates="related_c")
- Base.metadata.create_all(engine)
- Session = sessionmaker(bind=engine)
- session = Session()
- b11 = B(id=11)
- c111 = C(id=111)
- c111.related_b = b11
- a1 = A(id=1)
- b22 = B(id=22)
- a1.related_c = c111
- a1.related_b = b22
- session.add_all([b11, a1])
- session.commit()
- # verify by reading tables
- with engine.connect() as conn:
- print(conn.execute(sa.text("SELECT * FROM tbl_b")).fetchall())
- # [(11,), (22,)]
- print(conn.execute(sa.text("SELECT * FROM tbl_c")).fetchall())
- # [(111, 11)]
- print(conn.execute(sa.text("SELECT * FROM tbl_a")).fetchall())
- # [(1, 22, 111)]
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement