Skip to content

Instantly share code, notes, and snippets.

@blt

blt/Course.py

Created Jul 24, 2010
Embed
What would you like to do?
from sqlalchemy.orm import relationship, mapper
from tables import dept_tbl, inst_tbl, class_tbl, course_tbl
from tables import schedule_tbl, dept_inst, inst_course, meta
from tables import crs_edge_tbl, survey_tbl, avail_tbl
from objects import Dept, Inst, Class, Course, Schedule
from objects import Constraint, CourseEdge, Survey, Availability
mapper(Dept, dept_tbl, properties={
'classes': relationship(Class, backref='department',
cascade='all'),
'schedules': relationship(Schedule, backref='department',
cascade='all')
})
mapper(Inst, inst_tbl, properties={
'departments': relationship(Dept, secondary=dept_inst,
backref='instructors'),
'courses': relationship(Course,
secondary=inst_course,
backref='instructors')
})
mapper(Class, class_tbl, properties={
'offerings':relationship(Course, backref='Class', cascade='all')
})
mapper(Course, course_tbl)
mapper(CourseEdge, crs_edge_tbl, properties={
'lower_crs':relationship(Course,
primaryjoin=crs_edge_tbl.c.low_course_id==course_tbl.c.id,
backref='lower_edges', cascade='all', passive_deletes=True),
'higher_crs':relationship(Course,
primaryjoin=crs_edge_tbl.c.high_course_id==course_tbl.c.id,
backref='higher_edges', cascade='all', passive_deletes=True),
})
mapper(Schedule, schedule_tbl, properties={
'courses':relationship(Course, backref='schedule')
})
mapper(Survey, survey_tbl)
mapper(Availability, avail_tbl)
def create_all_tables(engine):
meta.create_all(engine)
import web, models
from util import crs_obj, dept_obj, dumpl, dumps, loads, get_insts
from util import cls_obj, schd_obj, get_courses
class courses:
def GET(self):
objs = web.ctx.orm.query(models.Course).all()
return dumpl(objs)
class CourseDept:
def GET(self, dept_id):
dept = dept_obj(dept_id)
if not dept:
return web.webapi.BadRequest()
objs = crs_obj(dept)
return dumpl(objs)
class CourseDeptYear:
def GET(self, dept_id, year):
dept = dept_obj(dept_id)
if not dept:
return web.webapi.BadRequest()
objs = crs_obj(dept, year)
return dumpl(objs)
class CourseDeptYearQrtr:
def GET(self, dept_id, year, qrtr):
dept = dept_obj(dept_id)
if not dept:
return web.webapi.BadRequest()
objs = crs_obj(dept, year, qrtr)
return dumpl(objs)
class CourseSec:
def GET(self, dept_id, year, qrtr, cls_id):
dept = dept_obj(dept_id)
if not dept:
return web.webapi.BadRequest()
objs = crs_obj(dept, year, qrtr, cls_id)
return dumpl(objs)
class Course:
def valid(self, dept_id, year, qrtr, cls_id, sec, obj):
try:
assert "class" in obj
assert tuple(obj["class"]["id"]) == (dept_id, int(cls_id))
assert "year" in obj
assert obj["year"] == int(year)
assert "quarter" in obj
assert obj["quarter"] == qrtr
assert "section" in obj
assert obj["section"] == int(sec)
assert "prerequisites" in obj
assert "times" in obj
assert len(obj["times"]) == 7
times = []
for day in ["monday", "tuesday", "wednesday", "thursday", "friday",
"saturday", "sunday"]:
assert day in obj["times"]
if (obj["times"][day] is None):
continue
else:
assert len(obj["times"][day]) == 2
times.append(obj["times"][day])
for time in times:
assert time == times[0]
except AssertionError:
return False
return True
def GET(self, dept_id, year, qrtr, cls_id, sec):
dept = dept_obj(dept_id)
if not dept:
return web.webapi.BadRequest()
obj = crs_obj(dept, year, qrtr, cls_id, sec)
return dumps(obj.hash())
def POST(self, dept_id, year, qrtr, cls_id, sec):
body = loads(web.data())
dept = dept_obj(dept_id)
if not dept:
return web.webapi.BadRequest()
cls = cls_obj(dept, cls_id)
if not cls:
return web.webapi.BadRequest()
obj = crs_obj(dept, year, qrtr, cls_id, sec)
if (not self.valid(dept_id, year, qrtr, cls_id, sec, body)) or obj:
return web.webapi.BadRequest()
# Schedule objects are implicit, we create them when
# they do not exist.
schd = schd_obj(dept, year, qrtr)
if not schd:
schd = models.Schedule(dept, year, qrtr)
web.ctx.orm.add(schd)
web.ctx.orm.commit()
insts = get_insts([i["id"] for i in body["instructors"]])
prereqs = get_courses(year, qrtr, body["prerequisites"])
days = [d for d, t in body["times"].iteritems() if t]
stime, etime = [t for t in body["times"].itervalues() if t][0]
c = models.Course(cls, schd, days, stime, etime, sec)
c.instructors = insts
web.ctx.orm.add(c)
for p in prereqs:
web.ctx.orm.add(c.add_prereq(p))
return web.webapi.created()
def PUT(self, dept_id, year, qrtr, cls_id, sec):
self.DELETE(dept_id, year, qrtr, cls_id, sec)
self.POST(dept_id, year, qrtr, cls_id, sec)
return web.webapi.Accepted()
def DELETE(self, dept_id, year, qrtr, cls_id, sec):
dept = dept_obj(dept_id)
if not dept:
return web.webapi.Accepted()
obj = crs_obj(dept, year, qrtr, cls_id, sec)
if obj:
web.ctx.orm.delete(obj)
return web.webapi.Accepted()
import time
import datetime
class Hasher(object):
def hash(self):
raise NotImplemented
class Dept(Hasher):
def __init__(self, abbrv, name):
self.abbrv = abbrv
self.name = name
def hash(self):
return {u"abbreviation": self.abbrv, u"name": self.name}
def __repr__(self):
return "<Dept('%(abbreviation)s','%(name)s')>" % \
self.hash()
class Inst(Hasher):
def __init__(self, id, fname, lname, depts=None):
self.psuid = id
self.fname = fname
self.lname = lname
if not depts:
self.departments = []
else:
self.departments = depts
def hash(self):
return {u'fname': self.fname, u'lname': self.lname,
u'id': self.psuid,
u'departments': [d.hash() for d in self.departments]}
def __repr__(self):
return "<Inst('%(id)s', '%(fname)s', '%(lname)s')>" % \
self.hash()
class Class(Hasher):
def __init__(self, cnum, name, dept_obj):
self.cnum = cnum
self.department = dept_obj
self.name = name
def hash(self):
return {u'name': self.name,
u'id': [self.department.abbrv, self.cnum]}
def __repr__(self):
return "<Class('%(name)s', '%(id)s'>" % \
self.hash()
class CourseEdge(object):
def __init__(self, c1, c2):
self.lower_crs = c1
self.higher_crs = c2
class Course(Hasher):
day_hash = {'monday' : int('00000001', 2),
'tuesday': int('00000010', 2),
'wednesday': int('00000100', 2),
'thursday': int('00001000', 2),
'friday': int('00010000', 2),
'saturday': int('00100000', 2),
'sunday': int('01000000', 2)}
def __init__(self, ClsObj, SchObj, days, stime, etime, section=None):
self.schedule = SchObj
self.Class = ClsObj
self.end_time = self._time_decode(etime)
self.start_time = self._time_decode(stime)
self.days = self._day_encode(days)
if section:
self.section = section
def _time_encode(self, t):
return t.isoformat()[:-3]
def _time_decode(self, tstr):
t_struct = time.strptime(tstr, "%H:%M")
return datetime.time(t_struct[3], t_struct[4])
def _day_encode(self, dlist):
dy = 0
for d in dlist:
dy |= self.day_hash[d]
return dy
def _day_decode(self, dint):
days = []
for k, v in self.day_hash.iteritems():
if dint & v:
days.append(k)
return days
def add_prereq(self, othercrs):
return CourseEdge(othercrs, self)
def satisfies(self):
return [x.higher_crs for x in self.lower_edges]
def prerequisites(self):
return [x.lower_crs for x in self.higher_edges]
def hash(self):
hours = [self._time_encode(self.start_time),
self._time_encode(self.end_time)]
times = dict([(d, None) for d in self.day_hash.keys()])
for d in self._day_decode(self.days):
times[d] = hours
return {u'times': times, u'class': self.Class.hash(),
u'section': self.section, 'year': self.schedule.year,
u'quarter': self.schedule.quarter,
u'prerequisites': [p.hash() for p in self.prerequisites()],
u'instructors': [i.hash() for i in self.instructors]}
def __repr__(self):
return str(self.hash())
class Schedule(Hasher):
def __init__(self, DeptObj, year, quarter):
self.department = DeptObj
self.year = year
self.quarter = quarter
def hash(self):
return {u'draft': self.draft,
u'department': self.department.hash(),
u'year': self.year, u'quarter': self.quarter,
u'courses': [c.hash() for c in self.courses]}
def __repr__(self):
return "<%(draft)s>" % self.hash()
class Constraint(Hasher):
def __init__(self, id, const):
self.id = id
self.const = const
def hash(self):
return {u'id' : self.id, u'constraints' : self.const}
def __repr__(self):
return "<Constraint('%(id)s', '%(const)s'>" % self.hash()
class Survey(Hasher):
def __init__(self, dept_obj, year, quarter, note=None):
self.department = dept_obj
self.year = year
self.quarter = quarter
self.note = note
def hash(self):
return {u'dept_id' : self.dept_id, u'year' : self.year,
u'quarter' : self.quarter, u'note' : self.note}
def __repr__(self):
st = "<Survey('%(dept_id)s', '%(inst_id)s', '%(year)s', '%(quarter)s'>"
return st % self.hash()
class Availability(Hasher):
def __init__(self, year, quarter, inst_obj, day, start, end):
self.instructor = inst_obj
self.year = year
self.quarter = quarter
self.day = day
self.start = start
self.end = end
def hash(self):
return {u'year' : self.year,
u'quarter' : self.quarter, u'inst_id' : self.inst_id,
u'day' : self.day, u'start' : self.start,
u'end' : self.end}
def __repr__(self):
st = "<Availability('%(day)s', '%(start)s', '%(end)s'>"
return st % self.hash()
from sqlalchemy import MetaData, Enum, Time, SmallInteger
from sqlalchemy import Table, Column, Integer, String
from sqlalchemy import ForeignKey, Boolean, UniqueConstraint
from sqlalchemy import ForeignKeyConstraint
meta = MetaData()
Quarter = Enum('spring', 'summer', 'fall', 'winter')
Day = Enum('M', 'Tu', 'W', 'Th', 'F')
#
# Straight-forward table definitions.
#
dept_tbl = Table('department', meta,
Column('id', Integer, primary_key=True, autoincrement=True),
Column('abbrv', String, nullable=False, unique=True),
Column('name', String, nullable=False)
)
inst_tbl = Table('instructor', meta,
Column('id', Integer, primary_key=True, autoincrement=True),
Column('psuid', String, nullable=False, unique=True),
Column('fname', String, nullable=False),
Column('lname', String, nullable=False),
)
class_tbl = Table('class', meta,
Column('id', Integer, primary_key=True, autoincrement=True),
Column('cnum', Integer, nullable=False),
Column('dept_id', Integer, ForeignKey('department.id'), nullable=False),
Column('name', String, nullable=False),
UniqueConstraint('cnum', 'dept_id')
)
schedule_tbl = Table('schedule', meta,
Column('id', Integer, primary_key=True, autoincrement=True),
Column('draft', Boolean, nullable=False, default=True),
Column('dept_id', Integer, ForeignKey('department.id'), nullable=False),
Column('year', SmallInteger, nullable=False),
Column('quarter', Quarter, nullable=False),
UniqueConstraint('dept_id', 'year', 'quarter')
)
course_tbl = Table('course', meta,
Column('id', Integer, primary_key=True, autoincrement=True),
Column('class_id', Integer, ForeignKey('class.id'),
nullable=False),
Column('sched_id', Integer, ForeignKey('schedule.id'),
nullable=False),
Column('section', Integer, nullable=False, default=1),
UniqueConstraint('class_id', 'sched_id', 'section'),
Column('days', SmallInteger, nullable=False),
Column('end_time', Time, nullable=False),
Column('start_time', Time, nullable=False),
)
crs_edge_tbl = Table('crs_edge', meta,
Column('low_course_id', Integer, primary_key=True),
Column('high_course_id', Integer, primary_key=True),
ForeignKeyConstraint(('low_course_id', 'high_course_id'),
('course.id', 'course.id'),
ondelete='cascade')
)
survey_tbl = Table('survey', meta,
Column('id', Integer, primary_key=True, autoincrement=True),
Column('dept_id', Integer, ForeignKey('department.id'), nullable=False),
Column('year', SmallInteger, nullable=False),
Column('quarter', Quarter, nullable=False),
Column('note', String),
UniqueConstraint('dept_id', 'year', 'quarter'))
avail_tbl = Table('availability', meta,
Column('id', Integer, primary_key=True, autoincrement=True),
Column('inst_id', Integer,
ForeignKey('instructor.id')),
Column('year', SmallInteger, nullable=False),
Column('quarter', Quarter, nullable=False),
Column('day', Day, nullable=False),
Column('start', String, nullable=False),
Column('end', String, nullable=False))
#
# Association tables.
#
dept_inst = Table('department_instructors', meta,
Column('dept_id', Integer, ForeignKey('department.id'), nullable=False),
Column('inst_id', Integer, ForeignKey('instructor.id'), nullable=False),
)
inst_course = Table('instructor_courses', meta,
Column('course_id', Integer, ForeignKey('course.id'),
nullable=False),
Column('inst_id', Integer, ForeignKey('instructor.id'), nullable=False),
)
import web
try:
from json import loads, dumps
except ImportError:
from simplejson import loads, dumps
import models
def dept_obj(dept):
return web.ctx.orm.query(models.Dept).filter_by(abbrv=dept).first()
def get_depts(dps):
ret = []
for d in dps:
dept = dept_obj(d)
assert dept
ret.append(dept)
return ret
def inst_obj(inst):
return web.ctx.orm.query(models.Inst).filter_by(psuid=inst).first()
def get_insts(insts):
ret = []
for i in insts:
inst = inst_obj(i)
assert inst
ret.append(inst)
return ret
def cls_obj(dept_obj, cls_id=None):
objs = web.ctx.orm.query(models.Class).filter_by(department=dept_obj)
if cls_id:
return objs.filter_by(cnum=cls_id).first()
return objs.all()
def crs_obj(dept_obj, year=None, qrtr=None, cls_id=None, sec=None):
objs = web.ctx.orm.query(models.Course)
objs = objs.join(models.Class).filter(models.Class.dept_id==dept_obj.id)
if year:
objs = objs.join(models.Schedule).filter(models.Schedule.year==year)
if qrtr:
assert year
objs = objs.filter(models.Schedule.quarter==qrtr)
if cls_id:
assert (year and qrtr)
objs = objs.filter(models.Class.cnum==cls_id)
if sec:
assert (year and qrtr and cls_id)
return objs.filter(models.Course.section==sec).first()
return objs.all()
def get_courses(year, qrtr, courses):
ret = []
for c in courses:
dept_id, cls_id = c
d_obj = dept_obj(dept_id)
assert d_obj
crs = crs_obj(d_obj, year, qrtr, cls_id)
assert crs
ret.extend(crs)
return ret
def schd_obj(dept_obj, year=None, qrtr=None):
objs = web.ctx.orm.query(models.Schedule).filter_by(dept_id=dept_obj.id)
if year:
objs = objs.filter_by(year=year)
if qrtr:
assert year
return objs.filter_by(quarter=qrtr).first()
return objs.all()
def dumpl(objs):
return dumps([x.hash() for x in objs])
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.