300:c9294e82e4f6
Anton Shestakov <av6@dwimlabs.net>, Sat, 23 Sep 2017 12:42:03 +0800
viewer: use order_by() explicitly, helps with keeping indices up-to-date

next change 301:508c24e708e0
previous change 296:7ae08f18f2b2

candolint/models.py

Permissions: -rw-r--r--

Other formats: Feeds:
from __future__ import absolute_import
import peewee as pw
from candolint import adapters
database = pw.SqliteDatabase(None, journal_mode='WAL')
class BaseModel(pw.Model):
class Meta:
database = database
@classmethod
def manual_upsert(cls, **kwargs):
try:
with cls._meta.database.atomic():
return cls.create(**kwargs), True
except pw.IntegrityError:
# a variation on https://github.com/coleifer/peewee/pull/681
query = []
compound_unique = set()
for field_names, unique in cls._meta.indexes:
if unique:
compound_unique.update(field_names)
for field_name, value in kwargs.items():
field = getattr(cls, field_name)
if field.unique or field.primary_key or field_name in compound_unique:
query.append(field == value)
instance = cls.get(*query)
for field_name, value in kwargs.items():
setattr(instance, field_name, value)
instance.save()
return instance, False
class Project(BaseModel):
url = pw.CharField(unique=True)
domain = pw.CharField()
user = pw.CharField(null=True)
name = pw.CharField()
class Meta:
indexes = (
(('domain', 'user', 'name'), True),
)
def get_url(self, request=None):
if request is not None:
base = '{}://{}'.format(request.protocol, request.host)
return base + self.get_url()
else:
user = self.user if self.user is not None else '-'
return '/{}/{}/{}'.format(self.domain, user, self.name)
def get_adapter(self):
if self.domain == 'bitbucket.org':
return adapters.BitbucketAdapter(self)
elif self.domain == 'github.com':
return adapters.GithubAdapter(self)
elif self.domain.startswith('hg.') or 'mercurial' in self.domain:
return adapters.HgwebAdapter(self)
else:
return adapters.HostingAdapter(self)
class Change(BaseModel):
rev = pw.IntegerField(null=True)
node = pw.CharField()
branch = pw.CharField()
date = pw.DateTimeField()
author = pw.CharField()
message = pw.CharField()
project = pw.ForeignKeyField(Project, related_name='changes')
class Meta:
indexes = (
(('node', 'project'), True),
)
class Check(BaseModel):
ordinal = pw.IntegerField()
errors = pw.IntegerField()
warnings = pw.IntegerField()
lines = pw.TextField()
success = pw.BooleanField()
started = pw.DateTimeField()
finished = pw.DateTimeField()
project = pw.ForeignKeyField(Project, related_name='checks')
change = pw.ForeignKeyField(Change, related_name='checks')
class Meta:
indexes = (
(('ordinal', 'project'), True),
(('project', 'finished', 'ordinal'), True)
)
def get_duration(self):
d = self.finished - self.started
minutes, seconds = divmod(int(d.total_seconds()), 60)
result = []
if minutes:
result.append('{} min'.format(minutes))
if seconds or not minutes:
result.append('{} sec'.format(seconds))
return ' '.join(result)
@classmethod
def get_next_ordinal(cls, project):
max_ordinal = (cls
.select(pw.fn.COALESCE(pw.fn.MAX(cls.ordinal), 0))
.where(cls.project == project))
return max_ordinal + 1
@classmethod
def get_light_fields(cls):
return [
field for field in cls._meta.fields.values()
if not isinstance(field, pw.TextField)
]