Source

pist / pist / models.py

Full commit
from collections import namedtuple
from datetime import datetime
from uuid import uuid4
import os

from mercurial import commands, hg

from constants import DIRNAME
from utils import get_dbc, safe_result, PistUI

CoreUser = namedtuple('User', [
    'id',
    'username',
    'method',
    'is_banned',
    'is_super',
    'date_created',
])

CoreRepo = namedtuple('Repo', [
    'id',
    'user_id',
    'date_created',
])

CoreRevision = namedtuple('Revision', ['id', 'date'])

class User(CoreUser):

    def __str__(self):
        return u'%s (%s)' % (self.username, self.method)

    @staticmethod
    def by_id(user_id):
        """Returns a User object based on the specified ID"""

        sql = 'SELECT * FROM `users` WHERE `id` = ? AND `is_banned` = 0'
        return safe_result(User, get_dbc().get(sql, user_id))

    @staticmethod
    def by_username(username, method):
        """Returns a User object based on their username and auth method"""

        sql = 'SELECT * FROM `users` WHERE `username` = ? AND `method` = ? AND `is_banned` = 0'
        return safe_result(User, get_dbc().get(sql, username, method))

    @staticmethod
    def create(username, method):
        """Creates and returns a new User object"""

        sql = 'INSERT INTO `users` (`username`, `method`, `date_created`) VALUES (?, ?, datetime())'
        last_id = get_dbc().execute(sql, username, method)
        user = User.by_id(last_id)

        return user

class Repo(CoreRepo):

    def __init__(self, *args, **kwargs):
        super(Repo, self).__init__(*args, **kwargs)
        self._ui = None
        self._repo = None

    def __str__(self):
        return self.hash

    def list(self):
        """Lists all files in the repo"""

        commands.manifest(self.ui, self.repo)
        return self.ui.flush()[0]

    def commit(self, message):
        """Commits any changes to the repo"""

        return commands.commit(self.ui, self.repo,
            message=message,
            addremove=True)

    def contents(self, file):
        """Returns the contents of a file in the repo"""

        return open(os.path.join(self.path, file), 'r').read()

    def revisions(self):
        """Returns a list of revisions for this repo"""

        commands.log(self.ui, self.repo, date=None, user=None, rev=None,
            template='{node} {date}')
        raw_revisions = self.ui.flush()[0]
        revisions = []
        for r in raw_revisions:
            if len(r.strip()):
                hash, date = r.split()
                date = datetime.fromtimestamp(float(date))
                revisions.append(Revision(hash, date))

        revisions.reverse()
        return revisions

    @property
    def revision(self):
        return self._revision

    @revision.setter
    def revision(self, rev):
        self._revision = rev
        return commands.update(self.ui, self.repo,
            rev=self._revision, clean=True)

    @property
    def hash(self):
        return self.id[:10]

    @property
    def path(self):
        return os.path.join(DIRNAME, 'repos', str(self.user_id), self.id)

    @property
    def ui(self):
        if not self._ui:
            self._ui = PistUI()

        return self._ui

    @property
    def repo(self):
        if not self._repo:
            self._repo = hg.repository(self.ui, self.path)

        return self._repo

    @staticmethod
    def by_id(hash):
        """Returns a repo based on its ID hash"""

        sql = 'SELECT * FROM `repos` WHERE `id` LIKE ?'
        return safe_result(Repo, get_dbc().get(sql, '%s%%' % hash))

    @staticmethod
    def by_user(user_id):
        """Returns a list of repos based on their owner's ID"""

        sql = 'SELECT * FROM `repos` WHERE `user_id` = ?'
        return [Repo(**row) for row in get_dbc().get(sql, user_id)]

    @staticmethod
    def create(user_id):
        """Creates a new repo for the specified user"""

        sql = 'INSERT INTO `repos` (`id`, `user_id`, `date_created`) VALUES (?, ?, datetime())'
        repo = Repo(uuid4().hex, user_id, datetime.now())
        get_dbc().execute(sql, repo.id, repo.user_id)

        # make the directories
        try:
            os.makedirs(repo.path)
        except OSError:
            pass

        # initialize a Mercurial repo
        commands.init(repo.ui, repo.path)

        return repo

class Revision(CoreRevision):

    def __str__(self):
        return self.hash

    @property
    def hash(self):
        return self.id[:10]