SQL fonctionne avec et renvoie des données tabulaires (ou des relations, si vous préférez y penser de cette façon, mais toutes les tables SQL ne sont pas des relations). Cela implique qu'une table imbriquée telle que décrite dans la question n'est pas une caractéristique si commune. Il existe des moyens de produire quelque chose de ce genre dans Postgresql, par exemple en utilisant des tableaux de JSON ou des composites, mais il est tout à fait possible de simplement récupérer des données tabulaires et d'effectuer l'imbrication dans l'application. Python a itertools.groupby()
, ce qui convient assez bien, étant donné les données triées.
L'erreur column "incoming.id" must appear in the GROUP BY clause... indique que les non-agrégats dans la liste de sélection, la clause ayant, etc. doivent apparaître dans le GROUP BY clause ou être utilisés dans un agrégat, de peur qu'ils n'aient éventuellement des valeurs indéterminées . En d'autres termes, la valeur devrait être sélectionnée à partir d'une seule ligne du groupe, car GROUP BY condense les lignes groupées en une seule ligne , et n'importe qui devinerait dans quelle rangée ils ont été choisis. L'implémentation peut le permettre, comme SQLite le fait et MySQL le faisait, mais la norme SQL l'interdit. L'exception à la règle est lorsqu'il existe une dépendance fonctionnelle
; le GROUP BY clause détermine les non-agrégats. Pensez à une jointure entre les tables A et B groupés par A la clé primaire de. Quelle que soit la ligne d'un groupe, le système choisirait les valeurs pour A à partir des colonnes de , elles seraient les mêmes puisque le regroupement a été effectué sur la base de la clé primaire.
Pour aborder l'approche générale prévue en 3 points, une façon serait de sélectionner une union d'entrants et de sortants, classés par leurs horodatages. Puisqu'il n'y a pas de hiérarchie d'héritage setup––comme il n'y en a peut-être même pas, je ne suis pas familier avec la comptabilité––un retour à l'utilisation des tuples de résultat Core et plain facilite les choses dans ce cas :
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)
Puis pour former la structure imbriquée itertools.groupby() est utilisé :
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]
Le résultat final est une liste de 2-tuples de date et une liste de dictionnaires d'entrées dans l'ordre croissant. Pas tout à fait la solution ORM, mais fait le travail. Un exemple :
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'}])]
Comme mentionné, Postgresql peut produire à peu près le même résultat qu'en utilisant un tableau de 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()
Si en fait Incoming et Outgoing peuvent être considérés comme des enfants d'une base commune, par exemple Entry , l'utilisation des unions peut être quelque peu automatisée avec héritage de table concrète
:
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
}
Malheureusement, en utilisant AbstractConcreteBase
nécessite un appel manuel à configure_mappers()
lorsque toutes les classes nécessaires ont été définies ; dans ce cas, la première possibilité est après avoir défini User , car Account en dépend par le biais de relations :
from sqlalchemy.orm import configure_mappers
configure_mappers()
Ensuite, afin de récupérer tous les Incoming et Outgoing dans une seule requête ORM polymorphe, utilisez Entry :
session.query(Entry).\
filter(Entry.accountID == accountID).\
order_by(Entry.timestamp).\
all()
et continuez à utiliser itertools.groupby() comme ci-dessus sur la liste résultante de Incoming et Outgoing .
