ORM related

---

---

SQLAlchemy

Inspect output raw SQL

# No .all() at the end
sql = db.query(TableA)

print(sql)

# Or 
print(sql.statement)

Join table in filter

db.query(TableA)
    .join(TableA.related_table_b_field_col)
    .filter(
        TableA.col1 == something,
        TableB.status == "active"
    )
    .all()

Use same subquery in LEFT & Right

• Let’s say we want to join 2 identical subquery

subquery = (
      db.query(TableA, TableB.col1, TableC.col11111)
      .select_from(TableA)
      .join(TableA.related_B_col1)
      .join(TableA.related_C_col1)
      .filter(
          TableB.status == "active",
          TableC.status == "disabled",
          TableA.id.not_in(some_list)
      )
      .subquery()
  )

# Set alias for main and join query
main_alias = aliased(subquery)
compare_alias = aliased(subquery)

# Build main query at last
main_q = (
    # Only select fields from LEFT query
    db.query(main_alias)
    .select_from(main_alias)
    .join(
        compare_alias,
        # Here is the manual on clause if necessary
        and_(
            main_alias.c.col1 == compare_alias.c.col2,
            main_alias.c.created_at < compare_alias.c.created_at
        ),
        # It is a LEFT JOIN
        isouter=True
    )
        # i.e. WHERE compare_alias.id is null
    .filter(compare_alias.c.id.is_(None))
)

# Execute SQL, empty array if not found
raw_res = main_q.all()
# Convert each rows: sqlalchemy.engine.row.Row into dict
main_res = [v._asdict() for v in raw_res]

Comments