models.py

import ast
import calendar
import datetime

import requests
from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
from django.contrib.contenttypes.models import ContentType
from django.db import connections, models
from django.db.models.functions import JSONObject
from django.utils.translation import gettext_lazy as _
from django_hosts.resolvers import reverse

from tracdb.models import Ticket
from tracdb.stats import get_trac_link

METRIC_PERIOD_INSTANT = "instant"
METRIC_PERIOD_DAILY = "daily"
METRIC_PERIOD_WEEKLY = "weekly"
METRIC_PERIOD_CHOICES = (
    (METRIC_PERIOD_INSTANT, _("Instant")),
    (METRIC_PERIOD_DAILY, _("Daily")),
    (METRIC_PERIOD_WEEKLY, _("Weekly")),
)


class Category(models.Model):
    name = models.CharField(max_length=300)
    position = models.PositiveSmallIntegerField(default=1)

    class Meta:
        verbose_name_plural = _("categories")

    def __str__(self):
        return self.name


class MetricQuerySet(models.QuerySet):
    def with_latest(self):
        """
        Annotate the queryset with a `latest` JSON object containing two keys:
            * `measurement` (int): the value of the most recent datum for that metric
            * `timestamp` (str): the timestamp of the most recent datum
        """
        data = Datum.objects.filter(
            content_type=self.model.content_type(),
            object_id=models.OuterRef("pk"),
        )
        jsonobj = JSONObject(
            measurement=models.F("measurement"),
            timestamp=models.F("timestamp"),
        )
        latest = models.Subquery(data.values_list(jsonobj).order_by("-timestamp")[:1])

        return self.annotate(latest=latest)

    def for_dashboard(self):
        """
        Return a queryset optimized for being displayed on the dashboard index
        page.
        """
        return (
            self.filter(show_on_dashboard=True).select_related("category").with_latest()
        )


class Metric(models.Model):
    name = models.CharField(max_length=300)
    slug = models.SlugField()
    category = models.ForeignKey(
        Category, blank=True, null=True, on_delete=models.SET_NULL
    )
    position = models.PositiveSmallIntegerField(default=1)
    data = GenericRelation("Datum")
    show_on_dashboard = models.BooleanField(default=True)
    show_sparkline = models.BooleanField(default=True)
    period = models.CharField(
        max_length=15, choices=METRIC_PERIOD_CHOICES, default=METRIC_PERIOD_INSTANT
    )
    unit = models.CharField(max_length=100)
    unit_plural = models.CharField(max_length=100)

    objects = MetricQuerySet.as_manager()

    class Meta:
        abstract = True

    def __str__(self):
        return self.name

    def get_absolute_url(self):
        return reverse("metric-detail", args=[self.slug], host="dashboard")

    @property
    def display_position(self):
        cat_position = -1 if self.category is None else self.category.position
        return cat_position, self.position

    def gather_data(self, since):
        """
        Gather all the data from this metric since a given date.

        Returns a list of (timestamp, value) tuples. The timestamp is a Unix
        timestamp, converted from localtime to UTC.
        """
        if self.period == METRIC_PERIOD_INSTANT:
            return self._gather_data_instant(since)
        elif self.period == METRIC_PERIOD_DAILY:
            return self._gather_data_periodic(since, "day")
        elif self.period == METRIC_PERIOD_WEEKLY:
            return self._gather_data_periodic(since, "week")
        else:
            raise ValueError("Unknown period: %s", self.period)

    def _gather_data_instant(self, since):
        """
        Gather data from an "instant" metric.

        Instant metrics change every time we measure them, so they're easy:
        just return every single measurement.
        """
        data = (
            self.data.filter(timestamp__gt=since)
            .order_by("timestamp")
            .values_list("timestamp", "measurement")
        )
        return [(calendar.timegm(t.timetuple()), m) for (t, m) in data]

    def _gather_data_periodic(self, since, period):
        """
        Gather data from "periodic" metrics.

        Period metrics are reset every day/week/month and count up as the period
        goes on. Think "commits today" or "new tickets this week".

        XXX I'm not completely sure how to deal with this since time zones wreak
        havoc, so there's right now a hard-coded offset which doesn't really
        scale but works for now.
        """
        OFFSET = "2 hours"  # HACK!

        c = connections["default"].cursor()
        c.execute(
            """SELECT
                        DATE_TRUNC(%s, timestamp - INTERVAL %s),
                        MAX(measurement)
                     FROM dashboard_datum
                     WHERE content_type_id = %s
                       AND object_id = %s
                       AND timestamp >= %s
                     GROUP BY 1;""",
            [period, OFFSET, self.content_type().id, self.id, since],
        )
        return [(calendar.timegm(t.timetuple()), float(m)) for (t, m) in c.fetchall()]

    @classmethod
    def content_type(cls):
        return ContentType.objects.get_for_model(cls)


class TracTicketMetric(Metric):
    query = models.TextField()

    def fetch(self):
        queryset = Ticket.objects.from_querystring(self.query)
        return queryset.count()

    def link(self):
        return get_trac_link(self.query)


class GithubItemCountMetric(Metric):
    """Example: https://api.github.com/repos/django/django/pulls?state=open"""

    api_url = models.URLField(max_length=1000)
    link_url = models.URLField(max_length=1000)

    def fetch(self):
        """
        Request the specified GitHub API URL with 100 items per page. Loop over
        the pages until no page left. Return total item count.
        """
        count = 0
        page = 1
        number_of_items_on_page = 101
        while number_of_items_on_page >= 100:
            r = requests.get(self.api_url, params={"page": page, "per_page": 100})
            number_of_items_on_page = len(r.json())
            count += number_of_items_on_page
            page += 1
        return count

    def link(self):
        return self.link_url


class GitHubSearchCountMetric(Metric):
    api_url = models.URLField(max_length=1000)
    link_url = models.URLField(max_length=1000)

    def fetch(self):
        """Request the specified GitHub API and return a total count."""
        today = datetime.date.today()
        if self.period == METRIC_PERIOD_WEEKLY:
            committer_date = ">%s" % (today - datetime.timedelta(weeks=1)).isoformat()
        else:
            committer_date = today.isoformat()
        r = requests.get(
            self.api_url
            + "?per_page=1&q=repo:django/django+committer-date:%s" % committer_date
        )
        data = r.json()
        return data["total_count"]


class JenkinsFailuresMetric(Metric):
    """
    Track failures of a job/build. Uses the Python flavor of the Jenkins REST
    API.
    """

    jenkins_root_url = models.URLField(
        verbose_name=_("Jenkins instance root URL"),
        max_length=1000,
        help_text=_("E.g. http://ci.djangoproject.com/"),
    )
    build_name = models.CharField(
        max_length=100,
        help_text=_("E.g. Django Python3"),
    )
    is_success_cnt = models.BooleanField(
        default=False,
        verbose_name=_("Should the metric be a value representing success ratio?"),
        help_text=_(
            "E.g. if there are 50 tests of which 30 are failing the value "
            "of this metric will be 20 (or 40%.)"
        ),
    )
    is_percentage = models.BooleanField(
        default=False,
        verbose_name=_("Should the metric be a percentage value?"),
        help_text=_(
            "E.g. if there are 50 tests of which 30 are failing the value of "
            "this metric will be 60%."
        ),
    )

    def urljoin(self, *parts):
        return "/".join(p.strip("/") for p in parts)

    def _fetch(self):
        """
        Actually get the values we are interested in by using the Jenkins REST
        API (https://wiki.jenkins-ci.org/display/JENKINS/Remote+access+API)
        """
        api_url = self.urljoin(self.link(), "api/python")
        job_desc = requests.get(api_url)
        job_dict = ast.literal_eval(job_desc.text)
        build_ptr_dict = job_dict["lastCompletedBuild"]
        build_url = self.urljoin(build_ptr_dict["url"], "api/python")
        build_desc = requests.get(build_url)
        build_dict = ast.literal_eval(build_desc.text)
        return (
            build_dict["actions"][4]["failCount"],
            build_dict["actions"][4]["totalCount"],
        )

    def _calculate(self, failures, total):
        """Calculate the metric value."""
        if self.is_success_cnt:
            value = total - failures
        else:
            value = failures
        if self.is_percentage:
            if not total:
                return 0
            value = (value * 100) / total
        return value

    def fetch(self):
        failures, total = self._fetch()
        return self._calculate(failures, total)

    def link(self):
        return self.urljoin(self.jenkins_root_url, "job", self.build_name)


class Datum(models.Model):
    metric = GenericForeignKey()
    content_type = models.ForeignKey(
        ContentType, related_name="+", on_delete=models.CASCADE
    )
    object_id = models.PositiveIntegerField()
    timestamp = models.DateTimeField(default=datetime.datetime.now)
    measurement = models.BigIntegerField()

    class Meta:
        ordering = ["-timestamp"]
        get_latest_by = "timestamp"
        verbose_name_plural = "data"

    def __str__(self):
        return f"{self.metric} at {self.timestamp}: {self.measurement}"