sqlalchemy one to many

sqlalchemy_one_to_many.py

import sqlalchemy as sa
import clickhouse_sqlalchemy as cs
from sqlalchemy.orm import relationship

DB_URL = "sqlite:///foo.db"

e = engine = sa.create_engine(DB_URL)
s = session = cs.make_session(engine)
m = metadata = sa.MetaData(bind=engine)
b = Base = cs.get_declarative_base(metadata=metadata)

manyOne = sa.Table(
    'manyOne', metadata,
    sa.Column('manyId', sa.Integer, sa.ForeignKey('Many.id')),
    sa.Column('oneId', sa.Integer, sa.ForeignKey('One.id'))
)


class Many(Base):
    __tablename__ = 'Many'
    id = sa.Column(sa.Integer, primary_key=True, autoincrement=True)
    name = sa.Column(sa.String, unique=True)


class One(Base):
    __tablename__ = 'One'
    id = sa.Column(sa.Integer, primary_key=True, autoincrement=True)
    name = sa.Column(sa.String, unique=True)
    manys = relationship('Many', secondary=manyOne, backref='One')


def init_data():
    objects = [
        Many(name='many1'),
        Many(name='many2')
    ]
    s.bulk_save_objects(objects)
    s.commit()
    r2 = One(name='One2')
    s1 = s.query(Many).filter(Many.name == 'many1').one()
    s2 = s.query(Many).filter(Many.name == 'many2').one()
    r2.manys = [s1, s2]
    s.add(r2)
    s.commit()
    s.flush()


def test_filter_One_by_manys():
    Ones = s.query(One).join(One.manys).filter(Many.name == 'many1').all()
    assert [r.name for r in Ones] == ['One1', 'One2']
    Ones = s.query(One).join(One.manys).filter(Many.name == 'many2').all()
    assert [r.name for r in Ones] == ['One2']


if __name__ == '__main__':
    Base.metadata.create_all()
    init_data()
    test_filter_One_by_manys()
    pass