# Also see
# http://moserei.de/2020/01/02/sqlalchemy-prevent-implicit-cross-join.html

import sqlalchemy as sa

engine = sa.create_engine("sqlite:///:memory:")
metadata = sa.MetaData()

a = sa.Table(
    "a",
    metadata,
    sa.Column("a_id", sa.Integer, primary_key=True, autoincrement=True),
    sa.Column("name", sa.String),
)
b = sa.Table(
    "b",
    metadata,
    sa.Column("b_id", sa.Integer, primary_key=True, autoincrement=True),
    sa.Column("a_id", sa.Integer, sa.ForeignKey(a.c.a_id)),
)


def create_select(additional_filters):
    return sa.select([a], whereclause=sa.and_(*additional_filters))


print(create_select([a.c.name == "Foo"]))
# SELECT a.a_id, a.name
# FROM a
# WHERE a.name = :name_1

print(create_select([a.c.name == "Foo", b.c.b_id == 1]))
# SELECT a.a_id, a.name
# FROM a, b
# WHERE a.name = :name_1 AND b.b_id = :b_id_1


def create_select_corrected(additional_filters):
    return sa.select([a], from_obj=a.join(b), whereclause=sa.and_(*additional_filters))


print(create_select_corrected([a.c.name == "Foo", b.c.b_id == 1]))
# SELECT a.a_id, a.name
# FROM a JOIN b ON a.a_id = b.a_id
# WHERE a.name = :name_1 AND b.b_id = :b_id_1


def before_execute(conn, clauseelement, multiparams, params):
    if (
        isinstance(clauseelement, sa.sql.selectable.Select)
        and len(clauseelement.froms) > 1
    ):
        raise RuntimeError("Cross join detected:\n{}".format(clauseelement))


sa.event.listen(engine, "before_execute", before_execute)

metadata.create_all(engine)

with engine.begin() as conn:
    # raises an exception
    conn.execute(create_select([a.c.name == "Foo", b.c.b_id == 1]))