sqlalchemy: express relationship which depends on len of collection child in joined entity?

Question

How do I express relationship which depends on len of collection child in joined entity?

In below example, parent entity is AlgoOrder. Child entity is Order. And PrivateTrade is child entity of Order.

AlgoOrder --> Order --> PrivateTrade

The problem I am having is with "orders_pending_private_trade_update".

class AlgoOrder(DbModel):
  __tablename__ = "algo_order"

  id = sa.Column(sa.Integer, primary_key=True)
  ... stuff ...

  # https://docs.sqlalchemy.org/en/14/orm/loading_relationships.html
  open_orders = orm.relation(Order, primaryjoin=and_(Order.algo_order_id == id, Order.status == 'OPEN'), lazy='select')
  orders_pending_private_trade_update = orm.relation(Order, primaryjoin=and_(Order.algo_order_id == id, , Order.status == 'CLOSED', len(Order.private_trades)==0), lazy='select')

  @property
  def pending_orders(self):
    return self.open_orders + self.orders_pending_private_trade_update


class Order(DbModel):
    __tablename__ = "order_hist"

    algo_order_id = sa.Column(sa.Integer, sa.ForeignKey("algo_order.id"))
    ... stiff ...

    private_trades = orm.relation(PrivateTrade, primaryjoin=and_(PrivateTrade.order_id == order_id))


class PrivateTrade(DbModel):
  __tablename__ = "private_trade"

  id = sa.Column(sa.Integer, primary_key=True)
  order_id = sa.Column(sa.String, sa.ForeignKey("order_hist.order_id"))

In particular, the error at "orders_pending_private_trade_update" was with "len" on Order.private_trades:

Exception has occurred: TypeError (note: full exception trace is shown but execution is paused at: _run_module_as_main) object of type 'InstrumentedAttribute' has no len()

So, I tried:

from sqlalchemy.sql.expression import func
orders_pending_private_trade_update = orm.relation(Order, primaryjoin=and_(Order.algo_order_id == id, Order.status == 'CLOSED', func.count(Order.private_trades)==0), lazy='select', viewonly=True)

But then error was "foreign key columns are present in neither the parent nor the child's mapped tables":

Can't determine relationship direction for relationship 'AlgoOrder.orders_pending_private_trade_update' - foreign key columns are present in neither the parent nor the child's mapped tables <class 'sqlalchemy.exc.ArgumentError'> Can't determine relationship direction for relationship 'AlgoOrder.orders_pending_private_trade_update' - foreign key columns are present in neither the parent nor the child's mapped tables

I checked my tables, I do have them:

op.create_table(
    'algo_order',
    sa.Column('id', sa.Integer(), primary_key=True),
    ...

op.create_table(
    'order_hist',
    sa.Column('id', sa.Integer(), primary_key=True),
    sa.Column('algo_order_id', sa.Integer, sa.ForeignKey("algo_order.id")),
    ...
op.create_table(
    'private_trade',
    sa.Column('id', sa.Integer(), primary_key=True),
    sa.Column('order_id', sa.String(), sa.ForeignKey("order_hist.order_id"))
    ...

Thanks in advance.

user3761555 · Answer 1 · 2022-07-12T01:01:12.037

I think I found it, but syntax pretty ugly: I used closed_order.session to do a new Query

import sqlalchemy as sa
import sqlalchemy.orm as orm
from sqlalchemy.sql.expression import func
import sqlalchemy.dialects.postgresql as psql
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.sql.expression import and_

class AlgoOrder(DbModel):
    __tablename__ = "algo_order"

    id = sa.Column(sa.Integer, primary_key=True)
    ... other stuff ...
    open_orders = orm.relation(Order, primaryjoin=and_(Order.algo_order_id == id, Order.status == 'OPEN'), lazy='select')
    closed_orders = orm.relation(Order, primaryjoin=and_(Order.algo_order_id == id, Order.status == 'CLOSED'), lazy='dynamic', viewonly=True)

    @property
    def orders_pending_private_trade_update(self):
        order_ids_with_no_private_trades = [ order.id for order in list(self.closed_orders.session.query(Order.id, func.count(PrivateTrade.id).label('count_private_trades')).join(PrivateTrade, isouter=True).group_by(Order.id).having(func.count(PrivateTrade.id) == 0).all())]
        orders_with_no_private_trades = self.closed_orders.session.query(Order).filter(Order.id.in_(order_ids_with_no_private_trades)).order_by(Order.id.desc()).limit(1000).all()
        return orders_with_no_private_trades

    @property
    def pending_orders(self):
        return list(self.open_orders) + list(self.orders_pending_private_trade_update)

I also don't like "first(100)" as an attempt to limit number of rows fetched. And how/when you dispose of the list to prevent memory leak? I think above approach is bad. Should use generator instead of returning list.

Essentially raw sql what I am looking for is a generator which returns below:

select
    order_id,
    cnt
    from (
        select
            order_hist.id,
            order_hist.order_id,
            count(private_trade.id) cnt
            from order_hist
            left join private_trade on private_trade.order_id = order_hist.order_id
            where order_hist.status in ('CLOSED', 'CANCELLED')
            group by order_hist.id, order_hist.order_id
    ) src
    where
    cnt=0

Any better way to do this? I think my solution shows the sqlalchemy syntax but it's computationally inefficient.

score 0 · Answer 2 · answered Jul 12 '22 at 01:49

Here's solution using generator instead to avoid MemoryError:

def order_hist_missing_private_trade_get(engine):
    order_hist_missing_private_trade_sql = '''
            select
                order_id,
                cnt
                from (
                    select
                        order_hist.id,
                        order_hist.order_id,
                        count(private_trade.id) cnt
                        from order_hist
                        left join private_trade on private_trade.order_id = order_hist.order_id
                        where order_hist.status in ('CLOSED', 'CANCELLED')
                        group by order_hist.id, order_hist.order_id
                ) src
                where
                cnt=0
            '''
    with engine.connect() as conn:
        # https://stackoverflow.com/questions/7389759/memory-efficient-built-in-sqlalchemy-iterator-generator
        conn.execution_options(stream_results=True)
        rs = conn.execute(order_hist_missing_private_trade_sql)
        while True:
            batch = rs.fetchmany(10000)
            for row in batch:
                order_id = row['order_id']
                yield order_id

Usage:

from sqlalchemy import create_engine

connstr : str = "postgresql://postgres:your_secret@localhost/postgres"
engine = create_engine(connstr)
generator = order_hist_missing_private_trade_get(engine)
while True:
    order_id = next(generator)
    print(f"order_id: {order_id}")

sqlalchemy: express relationship which depends on len of collection child in joined entity?

2 Answers2