The following code models a simple system that tracks the transfer of
construction tools between jobs. Equip (equipment) is transferred
between Jobs via Shipments.
Towards the end I attempt to map a class to a select statement in
order to make reporting simple. Instead of dealing with sql to do the
reporting, I wanted to map an object to a summarizing sql statement,
and create a sort of object model that covers most of the summarizing
I will need to do.
I can't figure out how to map an object to a select statement and
include a relation in the object. The code below should run in python
2.6:
from sqlalchemy import *
from sqlalchemy.orm import mapper, sessionmaker, relation
from sqlalchemy.sql import *
from datetime import date
# SA objects
db = create_engine('sqlite://', echo=True)
meta = MetaData()
session = sessionmaker(bind=db)()
# Table schema
job = Table('job', meta,
Column('id', Integer, primary_key=True),
Column('number', Integer))
equip = Table('equip', meta,
Column('id', Integer, primary_key=True),
Column('name', Unicode(255)))
equip_shipment = Table('equip_shipment', meta,
Column('id', Integer, primary_key=True),
Column('shipDate', Date),
Column('fromJobId', Integer, ForeignKey('job.id')),
Column('toJobId', Integer, ForeignKey('job.id')),
Column('isBroken', Boolean))
equip_shipment_item = Table('equip_shipment_item', meta,
Column('id', Integer, primary_key=True),
Column('shipmentId', Integer, ForeignKey
('equip_shipment.id')),
Column('equipId', Integer, ForeignKey('equip.id')),
Column('qty', Integer))
meta.create_all(db)
# Objects
class KeywordInitMixin(object):
'''Fills object's attributes with whatever keyword args were given
to init.
As an example, allows me to simply inherit from this class like
this:
class Test(KeywordInitMixin):
pass
...And then create objects like this:
t = Test(foo=1, bar='spam')
assert t.foo == 1
assert t.bar == 'spam'
'''
def __init__(self, **kwargs):
for attr in self.ATTRS:
if attr in kwargs:
setattr(self, attr, kwargs[attr])
else:
setattr(self, attr, None)
# Set any properties
for attr, val in kwargs.items():
# See if class has a property by this name
if (hasattr(self.__class__, attr) and
getattr(self.__class__, attr).__class__ is property):
setattr(self, attr, val)
def __repr__(self):
args = ['%s=%s' % (arg, val) for arg, val in
self.__dict__.items() if
arg in self.ATTRS and val]
args = ', '.join(args)
name = self.__class__.__name__
result = '%s(%s)' % (name, args)
return result
class Job(KeywordInitMixin):
ATTRS = ['number']
class Equip(KeywordInitMixin):
ATTRS = ['name']
class Shipment(KeywordInitMixin):
ATTRS = ['shipDate', 'fromJob', 'toJob', 'isBroken']
class ShipmentItem(KeywordInitMixin):
ATTRS = ['shipment', 'equip', 'qty']
# Map schema to objects
mapper(Job, job)
mapper(Equip, equip)
mapper(Shipment, equip_shipment,
properties={
'fromJob': relation(Job,
primaryjoin=equip_shipment.c.fromJobId==job.c.id),
'toJob': relation(Job,
primaryjoin=equip_shipment.c.toJobId==job.c.id),
}
)
mapper(ShipmentItem, equip_shipment_item,
properties={
'shipment': relation(Shipment, backref='items'),
'equip': relation(Equip)
}
)
# -------------------------------------
# Create some test data
# -------------------------------------
# Jobs
warehouse1 = Job(number=10001)
job1 = Job(number=11111)
job2 = Job(number=22222)
# Equipment
bClamps = Equip(name=u'Bridge Clamps')
cLocks = Equip(name=u'420 Channel Lock')
smallLock = Equip(name=u'Small 3210 Lock')
toolChest = Equip(name=u'Tool Chest')
# Add to orm
session.add_all([warehouse1, job1, job2, bClamps, cLocks])
# Ship tools to job 1
ship1 = Shipment(fromJob=warehouse1, toJob=job1, shipDate=date.today
())
ship1.items.append(ShipmentItem(qty=5, equip=bClamps))
# Transfer tools from job 1 to job 2
ship2 = Shipment(fromJob=job1, toJob=job2, shipDate=date.today())
ship2.items.append(ShipmentItem(qty=2, equip=bClamps))
# Job 1 returns some tools to the warehouse
ship3 = Shipment(fromJob=job1, toJob=warehouse1, shipDate=date.today
())
ship3.loadedBy = ship3.deliveredBy = 'jane doe'
ship3.items.append(ShipmentItem(qty=2, equip=smallLock))
# Add to orm
session.add_all([ship1, ship2, ship3])
# Job 1 breaks some tools
broken = Shipment(fromJob=job1, isBroken=True, shipDate=date.today())
broken.items.append(ShipmentItem(qty=1, equip=smallLock))
# Break more of same equip, but in different line item to test
aggregation
broken.items.append(ShipmentItem(qty=4, equip=smallLock))
# Job 2 breaks stuff too
broken2 = Shipment(fromJob=job2, isBroken=True, shipDate=date.today())
broken2.items.append(ShipmentItem(qty=1, equip=bClamps))
# Save
session.add_all([broken, broken2])
session.flush()
session.commit()
# -------------------------------------
# Create read-only reporting object that is based on a query that
# summarizes the equipment that has been broken by each job.
# -------------------------------------
class BrokenItem(object):
'''One row for each piece of equipment that a job has broken, and
the qty that has been broken.'''
pass
SELECT = [
equip_shipment.c.fromJobId,
equip.c.name.label('equip'),
func.sum(equip_shipment_item.c.qty).label('broken'),
]
FROM = [equip_shipment_item.join(equip_shipment).join(equip)]
sql = select(SELECT, from_obj=FROM)
sql = sql.where(equip_shipment.c.isBroken==True)
sql = sql.group_by(equip.c.name)
sql = sql.alias()
# Map to object
props = {
'job': relation(Job)
}
# I chose a primary key here because SA needs one
mapper(BrokenItem, sql, properties=props, primary_key=
[sql.c.fromJobId])
# ----------------------------------
# Test report object
# ----------------------------------
# Find what job 22222 has broke
# !! This causes an error because the table is joined in a way that
there
# is a row returned for each job !!
broke = session.query(BrokenItem).filter(Job.number==22222).one()
--~--~---------~--~----~------------~-------~--~----~
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to sqlalchemy@googlegroups.com
To unsubscribe from this group, send email to
sqlalchemy+unsubscr...@googlegroups.com
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en
-~----------~----~----~----~------~----~------~--~---
