Last active
July 3, 2018 20:35
-
-
Save disconnect3d/a0d38895f0fc5be515d1ad17ddd095a1 to your computer and use it in GitHub Desktop.
Wiechu from freenode irc asked to show how to do some things in sqlalchemy, so here it is.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# requires `pip install sqlalchemy` | |
import datetime | |
from sqlalchemy.ext.declarative import declarative_base | |
from sqlalchemy import create_engine, Column, String, Integer, ForeignKey, DateTime, join, func | |
from sqlalchemy.orm import Session | |
import sqlalchemy as sa | |
import logging | |
logging.basicConfig() | |
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO) | |
db_uri = 'sqlite:///db.sqlite' | |
engine = create_engine(db_uri) | |
meta = declarative_base(engine) | |
class Client(meta): | |
__tablename__ = 'clients' | |
id = Column(Integer, primary_key=True) | |
name = Column(String, nullable=False, unique=True) | |
def __str__(self): | |
return f'Client(id={self.id}, name={self.name})' | |
def __repr__(self): | |
return self.__str__() | |
class Invoice(meta): | |
__tablename__ = 'invoices' | |
id = Column(Integer, nullable=False, primary_key=True) | |
client_id = Column(Integer, ForeignKey(Client.id)) | |
date = Column(DateTime, nullable=False) | |
value = Column(Integer, nullable=False) | |
def __str__(self): | |
return f'Invoice(id={self.id}, client_id={self.client_id}, date={self.date}, value={self.value}' | |
def __repr__(self): | |
return self.__str__() | |
# Create all tables | |
meta.metadata.create_all() | |
session = Session(bind=engine, autoflush=True) | |
session.add_all( | |
Client(name=n) for n in ('wiechu', 'disconnect3d', 'somebody') | |
) | |
clients = session.query(Client).all() | |
print('--- clients ---') | |
for c in clients: | |
print(c) | |
print() | |
d0 = datetime.datetime.utcnow() | |
def gen_dt(offset): | |
return d0 - datetime.timedelta(days=offset) | |
session.add_all([ | |
Invoice(client_id=clients[0].id, date=d0, value=10), | |
Invoice(client_id=clients[0].id, date=d0, value=15), | |
Invoice(client_id=clients[0].id, date=gen_dt(1), value=1337), | |
Invoice(client_id=clients[0].id, date=gen_dt(1), value=200), | |
Invoice(client_id=clients[0].id, date=gen_dt(1), value=150), | |
Invoice(client_id=clients[1].id, date=gen_dt(1), value=50), | |
Invoice(client_id=clients[1].id, date=gen_dt(2), value=666), | |
Invoice(client_id=clients[1].id, date=gen_dt(2), value=12), | |
Invoice(client_id=clients[1].id, date=gen_dt(3), value=11), | |
Invoice(client_id=clients[1].id, date=gen_dt(3), value=12), | |
]) | |
invoices = session.query(Invoice).all() | |
print('--- invoices ---') | |
for i in invoices: | |
print(i) | |
print() | |
# The task is to get list of clients with their last invoice info (value, date) | |
# Some clients might not have any invoices and we want to see this too | |
""" | |
# Below is an equivalent of this SQL query: | |
SELECT c.name, nd.DATE, nd.VALUE FROM clients c | |
LEFT JOIN | |
( SELECT d.client_id client_id, d.DATE, n.VALUE | |
FROM | |
( SELECT client_id, MAX(DATE) DATE FROM invoices GROUP BY client_id) d | |
JOIN | |
( SELECT client_id, DATE, MAX(VALUE) VALUE FROM invoices GROUP BY client_id, DATE) n | |
ON | |
d.DATE=n.DATE AND d.client_id=n.client_id | |
) nd | |
ON | |
c.id=nd.client_id | |
--------------------- | |
From SqlAlchemy logging: | |
INFO:sqlalchemy.engine.base.Engine:SELECT clients.name AS clients_name, nd.date AS nd_date, nd.value AS nd_value | |
FROM clients LEFT OUTER JOIN (SELECT d.client_id AS client_id, d.date AS date, n.value AS value | |
FROM (SELECT invoices.client_id AS client_id, max(invoices.date) AS max_1, invoices.date AS date | |
FROM invoices GROUP BY invoices.client_id) AS d JOIN (SELECT invoices.client_id AS client_id, invoices.date AS date, max(invoices.value) AS value | |
FROM invoices GROUP BY invoices.client_id, invoices.date) AS n ON d.date = n.date AND d.client_id = n.client_id) AS nd ON clients.id = nd.client_id | |
""" | |
d = session.query(Invoice.client_id, func.max(Invoice.date), Invoice.date).group_by(Invoice.client_id).subquery('d') | |
n = session.query(Invoice.client_id, Invoice.date, func.max(Invoice.value).label('value')).group_by(Invoice.client_id, Invoice.date).subquery('n') | |
nd = session.query(d.c.client_id, d.c.date, n.c.value).join( | |
n, (d.c.date==n.c.date) & (d.c.client_id==n.c.client_id) | |
).subquery('nd') | |
q = session.query(Client.name, nd.c.date, nd.c.value).join( | |
nd, Client.id == nd.c.client_id, isouter=True # makes left outer join (which is the same as left join) | |
) | |
print('--- last invoices ---') | |
for args in list(q): | |
print(*args) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment