audrid / audrid / models.py

# 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 as_dict(self):
        return {
            "id" : self.id,
            "body" : json.loads(self.body),
            "type" : self.type,
            "deleted" : 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
        }
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.