'''
Based on the thread titled "filter relation based on column value"
at
http://groups.google.com/group/sqlalchemy/browse_thread/thread/0db7423d986ba543
and some others, I'm curious about how to better get my SqlA code
working
with "historical" (better term?) databases, where the relational state
of
the system depends on when the question is being asked.
Group membership, where both the members of the group, and the
"information"
about a group can change over time is shown in the example below.
'''
## for this simple example
## every employee can be in at most one group, duirng any time period
## groups can also change name over time, which doesn't doesn't
change
## the group id.
## employees don't change name during their tenure in a group
## Analysis Questions:
## 1. Show the composition of the group (with id=ii) at a given time,
## including name and all members
## 2. Show the history of a group over time, including name changes,
## membership changes
## 3. History for an employee, including when they change groups
'''
Questions:
1. Is there a "SQLAlchemical" way to write group_snapshot_ts into
a declarative class, such that the joins and loader respect the
time
constraints? (Read-only is fine as well on the loaded attributes)
a. eager_loader?
b. subclassing query?
2. (Secondary, off-topic)
Is there a constraint that shows that an employees time period in
a
group isn't *overlapping* their time period in another.
That is, that they are in only one group at once?
'''
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import CheckConstraint, ForeignKey, MetaData,
PrimaryKeyConstraint
from sqlalchemy import ForeignKeyConstraint, UniqueConstraint
from sqlalchemy import Table, Column, Integer, Boolean,Unicode, String
from sqlalchemy.orm import relation, backref
from sqlalchemy.orm import mapper
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.exc import NoResultFound
forever=2147483647 # maxint in sql system for int type
Base = declarative_base()
def db_setup(connstring='sqlite:///:memory:', echo=False):
engine = create_engine(connstring, echo=echo)
Session = sessionmaker(bind=engine, autoflush=False,
autocommit=False)
session = Session()
Base.metadata.bind = engine
Base.metadata.create_all()
return session, engine
class GroupInfo(Base):
__tablename__ = 'group_info'
group_id = Column(Integer, primary_key=True, nullable=False,
autoincrement=False)
group_name = Column(String, primary_key=True, nullable=False)
start = Column(Integer, nullable=False, primary_key=True,
autoincrement=False)
stop = Column(Integer, nullable=False, primary_key=True,
autoincrement=False)
__table_args__ = (
CheckConstraint('stop > start',name='ss_check'),
UniqueConstraint('group_id','start', name='u_start'),
UniqueConstraint('group_id','stop', name='u_stop'),
{}
)
class Membership(Base):
__tablename__ = 'membership'
group_id=Column(ForeignKey(GroupInfo.group_id, onupdate="cascade",
ondelete='cascade'))
employee_id = Column(Integer, primary_key=True, nullable=False)
start = Column(Integer, nullable=False, primary_key=True,
autoincrement=False)
stop = Column(Integer, nullable=False, primary_key=True,
autoincrement=False)
__table_args__ = (
CheckConstraint('stop > start',name='ss_check'),
UniqueConstraint
('group_id','employee_id','start',name='u_start'),
UniqueConstraint
('group_id','employee_id','stop',name='u_stop'),
{}
)
def group_snapshot_ts(session, groupid, ts):
GI = GroupInfo
G = session.query(GI).filter(GI.group_id==groupid)
G = G.filter(GI.start <= ts).filter(GI.stop >= ts)
try:
G = G.one()
except NoResultFound:
return None
M = Membership
members = session.query(M.employee_id).filter
(G.group_id==M.group_id)
members = members.filter(M.start <= ts).filter(M.stop >= ts)
G.members = [x[0] for x in members.all()]
return dict(id=G.group_id,name=G.group_name, members = G.members)
def demo(session):
## populate
gnames = ("group_id","group_name","start","stop")
for g in [(1,"group 1", 0,10),(1,"new group 1", 10,20),
(2,'group2', 3, forever)]:
session.add(GroupInfo( **(dict(zip(gnames,g)))))
session.flush()
mnames = ("group_id","employee_id","start","stop")
for e in [ (1,42,1,5),(2,42,5,100),(1,18,2,50)]:
session.add(Membership( **(dict(zip(mnames,e)))))
session.flush()
session.commit()
for (gid, ts) in (10,10), (1,3),(1,6),(1,11),(1,100),(2,1),(2,10):
print "Group %i, ts %i" % (gid,ts)
print group_snapshot_ts(session,gid,ts)
return None
## run it!
session, engine = db_setup()
demo(session)
--~--~---------~--~----~------------~-------~--~----~
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
-~----------~----~----~----~------~----~------~--~---
