SQLAlchemy:grupuj według dnia w wielu tabelach

SQL działa i zwraca dane tabelaryczne (lub relacje, jeśli wolisz myśleć w ten sposób, ale nie wszystkie tabele SQL są relacjami). Oznacza to, że zagnieżdżona tabela, taka jak przedstawiona w pytaniu, nie jest tak powszechną cechą. Istnieją sposoby na wytworzenie czegoś w tym rodzaju w Postgresql, na przykład za pomocą tablic JSON lub kompozytów, ale całkowicie możliwe jest po prostu pobranie danych tabelarycznych i wykonanie zagnieżdżenia w aplikacji. Python ma itertools.groupby() , co dość dobrze pasuje do rachunku, biorąc pod uwagę posortowane dane.

Błąd column "incoming.id" must appear in the GROUP BY clause... mówi, że nieagregaty na liście wyboru, zawierające klauzulę itp. muszą pojawić się w GROUP BY lub być używane w agregacie, aby nie miały prawdopodobnie nieokreślonych wartości . Innymi słowy wartość musiałaby zostać pobrana tylko z jakiegoś wiersza w grupie, ponieważ GROUP BY łączy zgrupowane wiersze w jeden wiersz i nikt nie zgadnie, z którego rzędu zostały wybrane. Implementacja może na to pozwolić, tak jak robił to SQLite i MySQL, ale standard SQL tego zabrania. Wyjątkiem od reguły jest sytuacja, gdy istnieje zależność funkcjonalna ; GROUP BY klauzula określa nieagregaty. Pomyśl o połączeniu między tabelami A i B pogrupowane według A klucz podstawowy. Bez względu na to, który wiersz w grupie system wybierze wartości dla A kolumny od, będą takie same, ponieważ grupowanie zostało wykonane na podstawie klucza podstawowego.

Aby zająć się 3 punktowym ogólnym zamierzonym podejściem, jednym ze sposobów byłoby wybranie unii przychodzących i wychodzących, uporządkowanych według ich znaczników czasu. Ponieważ nie ma hierarchii dziedziczenia konfiguracja – ponieważ może nawet nie być, nie jestem zaznajomiony z rachunkowością – powrót do używania krotek Core i zwykłych wyników ułatwia w tym przypadku:

incoming = select([literal('incoming').label('type'), Incoming.__table__]).\
    where(Incoming.accountID == accountID)

outgoing = select([literal('outgoing').label('type'), Outgoing.__table__]).\
    where(Outgoing.accountID == accountID)

all_entries = incoming.union(outgoing)
all_entries = all_entries.order_by(all_entries.c.timestamp)
all_entries = db_session.execute(all_entries)

Następnie, aby utworzyć zagnieżdżoną strukturę itertools.groupby() jest używany:

date_groups = groupby(all_entries, lambda ent: ent.timestamp.date())
date_groups = [(k, [dict(ent) for ent in g]) for k, g in date_groups]

Efektem końcowym jest lista 2 krotek dat oraz lista słowników haseł w porządku rosnącym. Niezupełnie rozwiązanie ORM, ale wykonuje zadanie. Przykład:

In [55]: session.add_all([Incoming(accountID=1, amount=1, description='incoming',
    ...:                           timestamp=datetime.utcnow() - timedelta(days=i))
    ...:                  for i in range(3)])
    ...:                  

In [56]: session.add_all([Outgoing(accountID=1, amount=2, description='outgoing',
    ...:                           timestamp=datetime.utcnow() - timedelta(days=i))
    ...:                  for i in range(3)])
    ...:                  

In [57]: session.commit()

In [58]: incoming = select([literal('incoming').label('type'), Incoming.__table__]).\
    ...:     where(Incoming.accountID == 1)
    ...: 
    ...: outgoing = select([literal('outgoing').label('type'), Outgoing.__table__]).\
    ...:     where(Outgoing.accountID == 1)
    ...: 
    ...: all_entries = incoming.union(outgoing)
    ...: all_entries = all_entries.order_by(all_entries.c.timestamp)
    ...: all_entries = db_session.execute(all_entries)

In [59]: date_groups = groupby(all_entries, lambda ent: ent.timestamp.date())
    ...: [(k, [dict(ent) for ent in g]) for k, g in date_groups]
Out[59]: 
[(datetime.date(2019, 9, 1),
  [{'accountID': 1,
    'amount': 1.0,
    'description': 'incoming',
    'id': 5,
    'timestamp': datetime.datetime(2019, 9, 1, 20, 33, 6, 101521),
    'type': 'incoming'},
   {'accountID': 1,
    'amount': 2.0,
    'description': 'outgoing',
    'id': 4,
    'timestamp': datetime.datetime(2019, 9, 1, 20, 33, 29, 420446),
    'type': 'outgoing'}]),
 (datetime.date(2019, 9, 2),
  [{'accountID': 1,
    'amount': 1.0,
    'description': 'incoming',
    'id': 4,
    'timestamp': datetime.datetime(2019, 9, 2, 20, 33, 6, 101495),
    'type': 'incoming'},
   {'accountID': 1,
    'amount': 2.0,
    'description': 'outgoing',
    'id': 3,
    'timestamp': datetime.datetime(2019, 9, 2, 20, 33, 29, 420419),
    'type': 'outgoing'}]),
 (datetime.date(2019, 9, 3),
  [{'accountID': 1,
    'amount': 1.0,
    'description': 'incoming',
    'id': 3,
    'timestamp': datetime.datetime(2019, 9, 3, 20, 33, 6, 101428),
    'type': 'incoming'},
   {'accountID': 1,
    'amount': 2.0,
    'description': 'outgoing',
    'id': 2,
    'timestamp': datetime.datetime(2019, 9, 3, 20, 33, 29, 420352),
    'type': 'outgoing'}])]

Jak wspomniano, Postgresql może wygenerować prawie taki sam wynik, jak przy użyciu tablicy JSON:

from sqlalchemy.dialects.postgresql import aggregate_order_by

incoming = select([literal('incoming').label('type'), Incoming.__table__]).\
    where(Incoming.accountID == accountID)

outgoing = select([literal('outgoing').label('type'), Outgoing.__table__]).\
    where(Outgoing.accountID == accountID)

all_entries = incoming.union(outgoing).alias('all_entries')

day = func.date_trunc('day', all_entries.c.timestamp)

stmt = select([day,
               func.array_agg(aggregate_order_by(
                   func.row_to_json(literal_column('all_entries.*')),
                   all_entries.c.timestamp))]).\
    group_by(day).\
    order_by(day)

db_session.execute(stmt).fetchall()

Jeśli w rzeczywistości Incoming i Outgoing mogą być traktowane jako dzieci o wspólnej podstawie, na przykład Entry , używanie unii może być nieco zautomatyzowane dzięki dziedziczeniu konkretnych tabel :

from sqlalchemy.ext.declarative import AbstractConcreteBase

class Entry(AbstractConcreteBase, Base):
    pass

class Incoming(Entry):
    __tablename__ = 'incoming'
    id          = Column(Integer,   primary_key=True)
    accountID   = Column(Integer,   ForeignKey('account.id'))
    amount      = Column(Float,     nullable=False)
    description = Column(Text,      nullable=False)
    timestamp   = Column(TIMESTAMP, nullable=False)
    account     = relationship("Account", back_populates="incomings")

    __mapper_args__ = {
        'polymorphic_identity': 'incoming',
        'concrete': True
    }

class Outgoing(Entry):
    __tablename__ = 'outgoing'
    id          = Column(Integer,   primary_key=True)
    accountID   = Column(Integer,   ForeignKey('account.id'))
    amount      = Column(Float,     nullable=False)
    description = Column(Text,      nullable=False)
    timestamp   = Column(TIMESTAMP, nullable=False)
    account     = relationship("Account", back_populates="outgoings")

    __mapper_args__ = {
        'polymorphic_identity': 'outgoing',
        'concrete': True
    }

Niestety używam AbstractConcreteBase wymaga ręcznego wywołania configure_mappers() kiedy wszystkie niezbędne klasy zostały zdefiniowane; w tym przypadku najwcześniejsza możliwość jest po zdefiniowaniu User , ponieważ Account zależy od tego poprzez relacje:

from sqlalchemy.orm import configure_mappers
configure_mappers()

Następnie, aby pobrać wszystkie Incoming i Outgoing w pojedynczym polimorficznym zapytaniu ORM użyj Entry :

session.query(Entry).\
    filter(Entry.accountID == accountID).\
    order_by(Entry.timestamp).\
    all()

i przejdź do użycia itertools.groupby() jak wyżej na liście wynikowej Incoming i Outgoing .