models.py

"""
Some models for pulling data from Trac.

Initially generated by inspectdb then modified heavily by hand, often by
consulting http://trac.edgewall.org/wiki/TracDev/DatabaseSchema.

A few notes on tables that're left out and why:

    * All the session and permission tables: they're just not needed.
    * Enum: I don't know what this is or what it's for.
    * NodeChange: Ditto.
"""

from datetime import date
from functools import reduce
from operator import and_, or_
from urllib.parse import parse_qs

from django.db import models

from .tractime import dayrange, time_property


class JSONBObjectAgg(models.Aggregate):
    function = "JSONB_OBJECT_AGG"  # PostgreSQL only.
    output_field = models.JSONField()


class TicketQuerySet(models.QuerySet):
    def with_custom(self):
        """
        Annotate the "custom" properties as a json blob.
        """
        return self.annotate(
            custom=JSONBObjectAgg("custom_fields__name", "custom_fields__value")
        )

    def from_querystring(self, querystring):
        parsed = parse_qs(querystring)
        model_fields = {f.name for f in self.model._meta.get_fields()}
        custom_lookup_required = False
        filter_kwargs, exclude_kwargs = {}, {}

        for field, (value,) in parsed.items():
            if field == "time":
                if value == "today..":
                    timestamp_range = dayrange(date.today(), 1)
                elif value == "thisweek..":
                    timestamp_range = dayrange(date.today(), 7)
                else:
                    raise ValueError(f"Unsupported time value {value}")

                filter_kwargs["_time__range"] = timestamp_range
                continue
            elif field not in model_fields:
                custom_lookup_required = True
                field = f"custom__{field}"
            if value.startswith("!"):
                exclude_kwargs[field] = value[1:]
            else:
                filter_kwargs[field] = value

        queryset = self
        if custom_lookup_required:
            queryset = queryset.with_custom()

        if exclude_kwargs:
            # negative values needed to be OR-ed for exclude
            q = reduce(or_, [models.Q(**{k: v}) for k, v in exclude_kwargs.items()])
            queryset = queryset.exclude(q)
        if filter_kwargs:
            # whereas positive values are AND-ed
            q = reduce(and_, [models.Q(**{k: v}) for k, v in filter_kwargs.items()])
            queryset = queryset.filter(q)

        return queryset


class Ticket(models.Model):
    id = models.AutoField(primary_key=True)
    type = models.TextField()

    _time = models.BigIntegerField(db_column="time", null=True)
    time = time_property("_time")

    _changetime = models.BigIntegerField(db_column="changetime", null=True)
    changetime = time_property("_changetime")

    component = models.ForeignKey(
        "Component",
        related_name="tickets",
        db_column="component",
        on_delete=models.DO_NOTHING,
        null=True,
    )
    severity = models.TextField()
    owner = models.TextField()
    reporter = models.TextField()
    cc = models.TextField()
    version = models.ForeignKey(
        "Version",
        related_name="tickets",
        db_column="version",
        on_delete=models.DO_NOTHING,
        null=True,
    )
    milestone = models.ForeignKey(
        "Milestone",
        related_name="tickets",
        db_column="milestone",
        on_delete=models.DO_NOTHING,
        null=True,
    )
    priority = models.TextField()
    status = models.TextField()
    resolution = models.TextField()
    summary = models.TextField()
    description = models.TextField()
    keywords = models.TextField()

    objects = TicketQuerySet.as_manager()

    class Meta:
        db_table = "ticket"
        managed = False

    def __str__(self):
        return f"#{self.id}: {self.summary}"


class TicketCustom(models.Model):
    pk = models.CompositePrimaryKey("ticket", "name")
    ticket = models.ForeignKey(
        Ticket,
        related_name="custom_fields",
        db_column="ticket",
        on_delete=models.DO_NOTHING,
    )
    name = models.TextField()
    value = models.TextField()

    class Meta:
        db_table = "ticket_custom"
        managed = False

    def __str__(self):
        return f"{self.name}: {self.value}"


class TicketChange(models.Model):
    pk = models.CompositePrimaryKey("ticket", "_time", "field")
    ticket = models.ForeignKey(
        Ticket,
        related_name="changes",
        db_column="ticket",
        on_delete=models.DO_NOTHING,
    )
    author = models.TextField()
    field = models.TextField()
    oldvalue = models.TextField()
    newvalue = models.TextField()

    _time = models.BigIntegerField(db_column="time")
    time = time_property("_time")

    class Meta:
        db_table = "ticket_change"
        managed = False
        ordering = ["_time"]

    def __str__(self):
        return f"#{self.ticket.id}: changed {self.field}"


class Component(models.Model):
    name = models.TextField(primary_key=True)
    owner = models.TextField()
    description = models.TextField()

    class Meta:
        db_table = "component"
        managed = False

    def __str__(self):
        return self.name


class Version(models.Model):
    name = models.TextField(primary_key=True)
    description = models.TextField()

    _time = models.BigIntegerField(db_column="time")
    time = time_property("_time")

    class Meta:
        db_table = "version"
        managed = False

    def __str__(self):
        return self.name


class Milestone(models.Model):
    name = models.TextField(primary_key=True)
    description = models.TextField()

    _due = models.BigIntegerField(db_column="_due")
    due = time_property("_due")

    _completed = models.BigIntegerField(db_column="_completed")
    completed = time_property("_completed")

    class Meta:
        db_table = "milestone"
        managed = False

    def __str__(self):
        return self.name


class SingleRepoRevisionManager(models.Manager):
    """
    Forces Revision to only query against a single repo, thus making
    Revision.rev behave something like a primary key.
    """

    def __init__(self, repo_id):
        self.repo_id = repo_id
        super().__init__()

    def get_queryset(self):
        qs = super().get_queryset()
        return qs.filter(repos=self.repo_id)


# Django's Trac uses a single repository with id 1 in the database
# These models will not work if another repository is ever added
# (but that seems unlikely at this point)
SINGLE_REPO_ID = 1


class Revision(models.Model):
    repos = models.IntegerField(default=SINGLE_REPO_ID)
    rev = models.TextField(primary_key=True)

    _time = models.BigIntegerField(db_column="time")
    time = time_property("_time")

    author = models.TextField()
    message = models.TextField()

    objects = SingleRepoRevisionManager(repo_id=SINGLE_REPO_ID)

    class Meta:
        db_table = "revision"
        managed = False

    def __str__(self):
        return "[{}] {}".format(self.rev, self.message.split("\n", 1)[0])


class Wiki(models.Model):
    pk = models.CompositePrimaryKey("name", "version")
    name = models.TextField()
    version = models.IntegerField()
    _time = models.BigIntegerField(db_column="time")
    time = time_property("_time")
    author = models.TextField()
    text = models.TextField()
    comment = models.TextField()
    readonly = models.IntegerField()

    class Meta:
        db_table = "wiki"
        managed = False

    def __str__(self):
        return f"{self.name} (v{self.version})"


class Attachment(models.Model):
    pk = models.CompositePrimaryKey("type", "id", "filename")
    type = models.TextField()
    id = models.TextField()
    filename = models.TextField()
    size = models.IntegerField()
    _time = models.BigIntegerField(db_column="time")
    time = time_property("_time")
    description = models.TextField()
    author = models.TextField()

    class Meta:
        db_table = "attachment"
        managed = False

    def __str__(self):
        attached_to = ("#%s" % self.id) if self.type == "ticket" else self.id
        return f"{self.filename} (on {attached_to})"