Skip to content

Instantly share code, notes, and snippets.

@disconnect3d
Last active July 3, 2018 20:35
Show Gist options
  • Save disconnect3d/a0d38895f0fc5be515d1ad17ddd095a1 to your computer and use it in GitHub Desktop.
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.
# 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