Source

audrid / audrid / models.py

Full commit
Martin Czygan 3bdd56d 







Martin Czygan 88ab47d 
Martin Czygan 3bdd56d 


Martin Czygan 88ab47d 































































Martin Czygan 3bdd56d 





























Martin Czygan 88ab47d 
Martin Czygan 3bdd56d 


Martin Czygan 88ab47d 
Martin Czygan 3bdd56d 



Martin Czygan 88ab47d 
Martin Czygan 3bdd56d 


Martin Czygan 88ab47d 

Martin Czygan 3bdd56d 



























Martin Czygan 88ab47d 
Martin Czygan 3bdd56d 













Martin Czygan 88ab47d 


Martin Czygan 3bdd56d 








Martin Czygan 88ab47d 










Martin Czygan 3bdd56d 









Martin Czygan 88ab47d 

Martin Czygan 3bdd56d 
Martin Czygan 88ab47d 

Martin Czygan 3bdd56d 
Martin Czygan 88ab47d 


Martin Czygan 3bdd56d 
Martin Czygan 88ab47d 
Martin Czygan 3bdd56d 

Martin Czygan 88ab47d 
Martin Czygan 3bdd56d 

Martin Czygan 88ab47d 










# coding: utf-8

"""
Database mappings.
"""

from audrid import (app, db)
from audrid.utils import random_id, random_id_fun
from audrid import validators as V
import datetime
import json

# def create_pool(pool=None, user_id=None):
#     """ Create a task (validate, persist and audit).
#     """
#     pool = V.pool(pool)
#     try:
#         doc = Document(id=pool['id'], body=json.dumps(pool))
#         log = Log(
#             document_id=pool['id'], 
#             body=json.dumps(pool), 
#             tag='pool_created', 
#             user_id=user_id, 
#             date=datetime.datetime.now())
#         db.session.add(doc)
#         db.session.flush()
#         db.session.add(log)
#         db.session.commit()
#     except Exception as exc:
#         db.session.rollback()
#         raise exc

# def update_pool(pool=None, user_id=None):
#     """ Update a pool.
#     """
#     pool = V.pool(pool)
#     doc = Document.query.filter_by(
#         id=pool['id']).filter_by(
#         deleted=False).first_or_404()
#     try:
#         doc.body = json.dumps(pool)
#         log = Log(
#             document_id=pool['id'], 
#             body=json.dumps(pool), 
#             tag='pool_updated', 
#             user_id=user_id, 
#             date=datetime.datetime.now())
#         db.session.add(doc)
#         db.session.flush()
#         db.session.add(log)
#         db.session.commit()
#     except Exception as exc:
#         # log error
#         db.session.rollback()

# def delete_pool(pid=None, user_id=None):
#     """ Delete a pool.
#     """
#     doc = Document.query.filter_by(
#         id=pid).filter_by(
#         deleted=False).first_or_404()
#     try:
#         doc.deleted = True
#         log = Log(
#             document_id=pid, 
#             body=json.dumps({}), 
#             tag='pool_deleted', 
#             user_id=user_id, 
#             date=datetime.datetime.now())
#         db.session.add(log)
#         db.session.commit()
#     except Exception as exc:
#         # log error
#         db.session.rollback()


class User(db.Model):
    """ A application user.
    """
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    password = db.Column(db.String(80), nullable=False)
    
    email = db.Column(db.String(120), unique=True, nullable=False)
    group = db.Column(db.String(120), nullable=False)

    active = db.Column(db.Boolean, nullable=False, default=True)

    def __init__(self, username=None, email=None, password=None, group='guest', active=True):
        self.username = username
        self.password = password
        self.email = email
        self.group = group
        self.active = active

    def __repr__(self):
        return '<User %s, %s>' % (self.username, self.email)


class Document(db.Model):
    """ A document. Can be a pool or an audit.
    """
    added_id = db.Column(db.Integer, primary_key=True)

    id = db.Column(db.String(16), unique=True, nullable=False)
    body = db.Column(db.LargeBinary, nullable=False)
    type = db.Column(db.String(32), nullable=False)

    deleted = db.Column(db.Boolean, nullable=False, default=False)

    def __init__(self, id=None, body=None, type=None, deleted=False):
        """ Last exit database.
        """
        self.id = id or random_id()
        self.body = body
        self.type = type
        self.deleted = deleted

    def __repr__(self):
        return '<Document %s, type=%s, length=%s, deleted=%s>' % (
            self.id, self.type, len(self.body), self.deleted)

    def to_python(self):
        """ Return the python representation (dictionary) of this document.
        """
        return json.loads(self.body)

class Log(db.Model):
    """ The main audit log.
    """
    added_id = db.Column(db.Integer, primary_key=True)

    document_id = db.Column(db.String(16), db.ForeignKey('document.id'), nullable=False)
    body = db.Column(db.LargeBinary)
    
    # what, when, who
    tag = db.Column(db.String(32), nullable=False)
    date = db.Column(db.DateTime, nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)

    def __init__(self, document_id=None, body=None, 
            date=None, tag=None, user_id=None):
        self.document_id = document_id
        self.body = body
        self.tag = tag
        self.date = date
        self.user_id = user_id

    def __repr__(self):
        return '<Log %s, %s, %s, %s>' % (self.added_id, self.document_id, self.tag, self.date)

class Exam(db.Model):
    """ Abstract exam.
    """
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(128), nullable=False, unique=True)
    description = db.Column(db.Text())

    # The root pool
    pool_id = db.Column(db.String(16), db.ForeignKey('document.id'), nullable=False)
    
    public = db.Column(db.Boolean, nullable=False, default=True)    
    secret_key = db.Column(db.String(32), nullable=False, default=random_id_fun(length=32))

    # deleted 
    deleted = db.Column(db.Boolean, nullable=False, default=False)

    def __init__(self, name=None, description=None, public=True, 
        pool_id=None, secret_key=None):
        self.name = name
        self.description = description
        self.public = public
        self.pool_id = pool_id
        if self.secret_key:
            self.secret_key = secret_key

    def as_dict(self):
        return {
            "id" : self.id,
            "name" : self.name,
            "description" : self.description,
            "pool_id" : self.pool_id,
            "public" : self.public,
            "secret_key" : self.secret_key,
            "deleted" : self.deleted
        }

    def __repr__(self):
        return '<Exam %s, %s>' % (self.id, self.name)

class Audit(db.Model):
    """ Connects a user and an exam. E.g. 'Geography 1' to user 12.
    """
    id = db.Column(db.Integer, primary_key=True)

    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
    exam_id = db.Column(db.Integer, db.ForeignKey('exam.id'), nullable=False)
    # accumulate answers here
    trial_id = db.Column(db.String(16), db.ForeignKey('document.id'), nullable=False)

    # start, end date and duration
    started = db.Column(db.DateTime(), nullable=True)
    finished = db.Column(db.DateTime(), nullable=True, default=None)
    duration = db.Column(db.Integer, nullable=False, default=60) # in minutes

    status = db.Column(db.String(32), nullable=False, default='registered')

    def __init__(self, user_id=None, exam_id=None, trial_id=None):
        self.user_id = user_id
        self.exam_id = exam_id
        self.trial_id = trial_id

    def __repr__(self):
        return '<Audit %s, %s, %s>' % (self.id, self.status, self.trial_id)

    def as_dict(self):
        return {
            "user_id" : self.user_id,
            "exam_id" : self.exam_id,
            "trial_id" : self.trial_id,
            "started" : self.started,
            "finished" : self.finished,
            "duration" : self.duration,
            "status" : self.status
        }