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