backlog/main/models.py

import datetime
import uuid

from django.contrib.auth.models import User
from django.contrib.contenttypes.fields import GenericForeignKey, \
    GenericRelation
from django.contrib.contenttypes.models import ContentType
from django.core import serializers
from django.db import models, connection
from django.db.models.aggregates import Sum
from martor.models import MartorField
from notifications.signals import notify


class Member(models.Model):
    user = models.OneToOneField(User, on_delete=models.CASCADE)
    trigram = models.CharField(max_length=5)

    def __str__(self):
        return self.trigram

class BaseModel(models.Model):
    created = models.DateTimeField(auto_now_add=True)
    updated = models.DateTimeField(auto_now=True)
    objects = models.Manager()
    uuid = models.UUIDField(default=uuid.uuid4, editable=False)
    
    class Meta:
        abstract = True

    def get_content_type(self):
        return ContentType.objects.get_for_model(self)

    def model_name(self):
        try:
            return self._meta.verbose_name
        except AttributeError:
            return ""

    def to_json(self):
        return serializers.serialize('json', [ self, ])


class Comment(BaseModel):
    class Meta:
        verbose_name = "commentaire"
        verbose_name_plural = "commentaires"
        ordering = ('-created', )
        
    def __str__(self):
        return "De {}, le {:%Y-%m-%d}: {}".format(self.author.username, self.created, self.content[:30])
                
    content_type = models.ForeignKey(ContentType,on_delete=models.CASCADE)
    object_id = models.PositiveIntegerField()
    content_object = GenericForeignKey('content_type', 'object_id')
    
    content = MartorField(blank=False, default="", verbose_name="Commentaire")
    author = models.ForeignKey(User, 
                               on_delete=models.PROTECT, 
                               verbose_name="Auteur")
    created = models.DateTimeField(auto_now_add=True)
    
class Project(BaseModel):
    class Meta:
        verbose_name = "projet"
        verbose_name_plural = "projets"
        
    name = models.CharField(max_length=200)
    description = MartorField(blank=True, default="", verbose_name="Description")
    color = models.CharField(max_length=7, default="#f6755e")
    
    def __str__(self):
        return self.name
    
class Epic(BaseModel):
    class Meta:
        verbose_name = "epic"
        verbose_name_plural = "epics"
        ordering = ('-value', )
        
    SIZES = (("XXS","XXS (< 1 sprints)"), 
             ("XS", "XS (1 sprint)"), 
             ("S", "S (1-2 sprints)"), 
             ("M", "M (2-3 sprints)"), 
             ("L", "L (3-4 sprints)"), 
             ("XL", "XL (4-6 sprints)"), 
             ("XXL", "XXL  (> 6 sprints)"))
    
    name = models.CharField(max_length=200, default="", verbose_name="Nom")
    size = models.CharField(max_length=10, default="M", choices=SIZES, verbose_name="Taille")
    value = models.IntegerField(default=0, verbose_name="Valeur")
    description = MartorField(blank=True, default="", verbose_name="Description")
    project = models.ForeignKey(Project, on_delete=models.PROTECT, null=True, verbose_name="Projet")
    closed = models.BooleanField(default=False, verbose_name="Clôturée")
    comments = GenericRelation(Comment)
    
    def __str__(self):
        return self.name
    
    def nb_stories(self):
        return len(self.stories.all())
    
    def nb_closed_stories(self):
        return len(self.stories.filter(closed=True))
    
    def nb_active_stories(self):
        return len([story for story in self.stories.filter(closed=False) if story.running()])
    
    def contributors(self):
        qry = User.objects.raw("""SELECT DISTINCT auth_user.* 
                                    FROM ((auth_user INNER JOIN main_story_assignees ON auth_user.id = main_story_assignees.user_id) 
                                    INNER JOIN main_story ON main_story_assignees.story_id = main_story.id)
                                    WHERE main_story.epic_id = {};
                                    """.format(self.id))
        return list(qry)
    
    def contributors_str(self):
        return ", ".join([c.member.trigram for c in self.contributors()])
    
    def close(self):
        self.closed = True
        self.save()
        
    def reopen(self):
        self.closed = False
        self.save()
        
class Sprint(BaseModel):
    class Meta:
        verbose_name = "sprint"
        verbose_name_plural = "sprints"
        ordering = ('-date_start', )
    date_start = models.DateField()
    date_end = models.DateField()
    closed = models.BooleanField(default=False, verbose_name="Terminé")
    retro = MartorField(blank=True, default="", verbose_name="Bilan / Rétrospective")
        
    def __str__(self):
        return "Sprint #{} ({:%d/%m/%Y} > {:%d/%m/%Y})".format(self.id, self.date_start, self.date_end)
    
    def running(self):
        return self.date_start <= datetime.date.today() <= self.date_end and not self.closed
    
    def nb_stories(self):
        return self.stories.count()
    
    def planned_velocity(self):
        total =  self.stories.aggregate(Sum('weight'))['weight__sum']
        return total if total is not None else "NA"
    
    def real_velocity(self):
        if datetime.date.today() < self.date_start:
            return "NA"
        sql = """SELECT SUM(main_story.weight) as vel
                FROM main_story 
                    INNER JOIN (SELECT story_id, max(sprint_id) as sprint_id
                                FROM main_story_sprints 
                                GROUP BY story_id) as last_sprints 
                    ON main_story.id = last_sprints.story_id
                WHERE last_sprints.sprint_id = {} AND closed = True;
            """.format(self.id)
        with connection.cursor() as cursor:
            cursor.execute(sql)
            row = cursor.fetchone()
        return row[0] if row[0] else 0

    @classmethod
    def current(cls):
        """ the current sprint is the first non-closed sprint """
        try:
            return Sprint.objects.filter(closed = False).order_by('date_start')[0]
        except IndexError:
            return None

    @classmethod
    def next(cls):
        try:
            return Sprint.objects.filter(closed = False).order_by('date_start')[1]
        except IndexError:
            return None

class Story(BaseModel):
    class Meta:
        verbose_name = "story"
        verbose_name_plural = "stories"
        ordering = ('closed', '-updated')
        
    WEIGHTS = ((None, '------'), (1, 1),(2, 2),(3, 3),(5, 5),(8, 8),(13, 13),(21, 21))
        
    epic = models.ForeignKey(Epic, 
                             on_delete=models.PROTECT, 
                             null=True, blank=True,
                             related_name="stories", 
                             verbose_name="Epic")
    name = models.CharField(max_length=200, default="", verbose_name="Nom")
    weight = models.IntegerField(null=True, choices=WEIGHTS, verbose_name="Poids")
    description = MartorField(blank=True, default="", verbose_name="Description")
    closed = models.BooleanField(default=False, verbose_name="Clôturée")
    author = models.ForeignKey(User, 
                               on_delete=models.PROTECT, 
                               null=True, blank=True,
                               related_name="stories",
                               related_query_name="story", 
                               verbose_name="Auteur")
    
    assignees = models.ManyToManyField(User, 
                                       blank=True,
                                       related_name="assigned", 
                                       verbose_name="Assignés")
    
    sprints = models.ManyToManyField(Sprint, 
                                       blank=True,
                                       related_name="stories",
                                       related_query_name="story", 
                                       verbose_name="Sprints")
    comments = GenericRelation(Comment)
    
    def __str__(self):
        return self.name
                
    def running(self):
        for sprint in self.sprints.all():
            if sprint.running():
                return True
        return False

    def contributors(self):
        contributors = list(self.assignees.all())
        if not self.author in contributors:
            contributors += [self.author]
        return contributors

    def close(self):
        self.closed = True
        self.save()

    def reopen(self):
        self.closed = False
        self.save()