SQLAlchemy:nhóm theo ngày trên nhiều bảng

SQL hoạt động với và trả về dữ liệu dạng bảng (hoặc quan hệ, nếu bạn thích nghĩ theo cách đó, nhưng không phải tất cả các bảng SQL đều là quan hệ). Điều này ngụ ý rằng một bảng lồng nhau như được mô tả trong câu hỏi không phải là một tính năng phổ biến. Có nhiều cách để tạo ra một thứ gì đó thuộc loại này trong Postgresql, chẳng hạn như sử dụng mảng JSON hoặc vật liệu tổng hợp, nhưng bạn hoàn toàn có thể chỉ tìm nạp dữ liệu dạng bảng và thực hiện lồng ghép trong ứng dụng. Python có itertools.groupby() , khá phù hợp với hóa đơn, với dữ liệu được sắp xếp.

Lỗi column "incoming.id" must appear in the GROUP BY clause... đang nói rằng các tập hợp không phải trong danh sách chọn, mệnh đề có, v.v. phải xuất hiện trong GROUP BY hoặc được sử dụng trong một mệnh đề tổng hợp, vì sợ rằng chúng có thể có giá trị không xác định . Nói cách khác, giá trị sẽ phải được chọn chỉ từ một số hàng trong nhóm, bởi vì GROUP BY cô đọng các hàng đã nhóm thành một hàng duy nhất , và mọi người sẽ đoán họ được chọn từ hàng nào. Việc triển khai có thể cho phép điều này, giống như SQLite và MySQL đã từng làm, nhưng tiêu chuẩn SQL cấm như vậy. Ngoại lệ đối với quy tắc là khi có phụ thuộc hàm ; GROUP BY mệnh đề xác định các phi tập hợp. Hãy nghĩ về phép nối giữa các bảng A và B được nhóm bởi A khóa chính của. Bất kể hàng nào trong nhóm, hệ thống sẽ chọn các giá trị cho A của các cột từ, chúng sẽ giống nhau vì việc nhóm được thực hiện dựa trên khóa chính.

Để giải quyết phương pháp tiếp cận dự định chung 3 điểm, một cách sẽ là chọn một nhóm đến và đi, được sắp xếp theo dấu thời gian của chúng. Vì không có phân cấp kế thừa thiết lập –– vì thậm chí có thể không có, tôi không quen thuộc với kế toán –– việc quay lại sử dụng Bộ giá trị cốt lõi và kết quả đơn giản giúp mọi thứ dễ dàng hơn trong trường hợp này:

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)

Sau đó, để tạo cấu trúc lồng nhau itertools.groupby() được sử dụng:

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]

Kết quả cuối cùng là một danh sách gồm 2 bộ ngày tháng và danh sách các từ điển của các mục nhập theo thứ tự tăng dần. Không hoàn toàn là giải pháp ORM, nhưng hoàn thành công việc. Ví dụ:

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'}])]

Như đã đề cập, Postgresql có thể tạo ra khá nhiều kết quả giống như đang sử dụng một mảng 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()

Nếu thực tế thì Incoming và Outgoing có thể được coi là con của một cơ sở chung, ví dụ:Entry , việc sử dụng liên hợp có thể hơi tự động với kế thừa bảng cụ thể :

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
    }

Rất tiếc khi sử dụng AbstractConcreteBase yêu cầu lệnh gọi thủ công tới configure_mappers() khi tất cả các lớp cần thiết đã được xác định; trong trường hợp này, khả năng sớm nhất là sau khi xác định User , bởi vì Account phụ thuộc vào nó thông qua các mối quan hệ:

from sqlalchemy.orm import configure_mappers
configure_mappers()

Sau đó, để tìm nạp tất cả Incoming và Outgoing trong một truy vấn ORM đa hình, hãy sử dụng Entry :

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

và tiếp tục sử dụng itertools.groupby() như trên trong danh sách kết quả của Incoming và Outgoing .