models.py

from __future__ import annotations
import json
import datetime
from django.utils.translation import gettext_lazy as _
from django.db import models
from django.forms.models import model_to_dict
from django.contrib.auth import get_user_model
from . import providers


class IdentityManager(models.Manager):
    def create(self, **kw):
        """
        Create a new process and all its requirements
        """
        audit_author = kw.pop("audit_author", None)
        audit_notes = kw.pop("audit_notes", None)
        audit_skip = kw.pop("audit_skip", False)

        obj = self.model(**kw)
        obj.save(using=self._db, audit_author=audit_author, audit_notes=audit_notes, audit_skip=audit_skip)
        return obj


class Identity(models.Model):
    """
    Identity for a user in a remote user database
    """
    class Meta:
        unique_together = ["issuer", "subject"]

    person = models.ForeignKey(get_user_model(), related_name="identities", null=True, on_delete=models.CASCADE)
    issuer = models.CharField(max_length=512, help_text=_("identifier of the user database that manages this identity"))
    subject = models.CharField(max_length=512, help_text=_("identifier of the person in the user database"))
    last_used = models.DateField(auto_now=True, help_text=_("last time this identity has been used"))
    profile = models.CharField(max_length=1024, blank=True, help_text=_("URL profile information"))
    picture = models.CharField(max_length=1024, blank=True, help_text=_("URL to profile picture"))
    fullname = models.CharField(_("full name"), max_length=255, blank=True)
    username = models.CharField(max_length=512, blank=True, help_text=_("username in the remote user database"))

    objects = IdentityManager()

    def __str__(self):
        if self.username:
            return f"{self.person if self.person else '-'}:{self.issuer}:{self.subject} (aka {self.username})"
        else:
            return f"{self.person if self.person else '-'}:{self.issuer}:{self.subject} (no remote username)"

    def get_provider(self) -> providers.Provider:
        """
        Return the Provider information for this Identity
        """
        return providers.get(self.issuer)

    def update(
            self,
            profile=None, picture=None, fullname=None, username=None,
            audit_author=None, audit_notes="", audit_skip=False):
        """
        Update profile information with extra data that may be provided in
        import / login flows. Save only if needed.
        """
        changed = False

        if profile is not None and self.profile != profile:
            self.profile = profile
            changed = True

        if picture is not None and self.picture != picture:
            self.picture = picture
            changed = True

        if fullname is not None and self.fullname != fullname:
            self.fullname = fullname
            changed = True

        if username is not None and self.username != username:
            self.username = username
            changed = True

        if changed:
            self.save(
                audit_author=audit_author, audit_notes=audit_notes,
                audit_skip=audit_skip)

        return changed

    def save(self, *args, **kw):
        """
        Save, and add an entry to the Identity audit log.

        Extra arguments that can be passed:

            audit_author: user model instance of the person doing the change
            audit_notes: free form text annotations for this change
            audit_skip: skip audit logging, used only for tests

        """
        # Extract our own arguments, so that they are not passed to django
        author = kw.pop("audit_author", None)
        notes = kw.pop("audit_notes", "")
        audit_skip = kw.pop("audit_skip", False)

        if audit_skip:
            changes = None
        else:
            # Get the previous version of the object, so that diff() can
            # compute differences
            if self.pk:
                old = Identity.objects.get(pk=self.pk)
            else:
                old = None

            changes = Identity.diff(old, self)
            if changes and not author:
                raise RuntimeError("Cannot modify an Identity instance without providing Author information")

        # Perform the save; if we are creating a new identity, this will also
        # fill in the id/pk field, so that IdentityAuditLog can link to us
        super().save(*args, **kw)

        # Finally, create the identity audit log entry
        if changes:
            IdentityAuditLog.objects.create(
                    identity=self, author=author, notes=notes, changes=IdentityAuditLog.serialize_changes(changes))

    @classmethod
    def diff(cls, old_obj: "Identity", new_obj: "Identity"):
        """
        Compute the changes between two different instances of a Person model
        """
        def to_dict(obj):
            res = model_to_dict(obj, exclude=["id", "last_used", "person"])
            if obj.person is None:
                res["person"] = None
            else:
                res["person"] = str(obj.person)
            return res

        changes = {}
        if old_obj is None:
            for k, nv in list(to_dict(new_obj).items()):
                changes[k] = [None, nv]
        else:
            old = to_dict(old_obj)
            new = to_dict(new_obj)
            for k, nv in list(new.items()):
                ov = old.get(k, None)
                # Also ignore changes like None -> ""
                if ov != nv and (ov or nv):
                    changes[k] = [ov, nv]
        return changes


class IdentityAuditLog(models.Model):
    identity = models.ForeignKey(Identity, related_name="audit_log", on_delete=models.CASCADE)
    logdate = models.DateTimeField(null=False, auto_now_add=True)
    author = models.ForeignKey(get_user_model(), related_name="+", null=False, on_delete=models.CASCADE)
    notes = models.TextField(null=False, default="")
    changes = models.TextField(null=False, default="{}")

    def __str__(self):
        return "{:%Y-%m-%d %H:%S}:{}: {}:{}".format(
                self.logdate, self.identity, self.author, self.notes)

    @classmethod
    def serialize_changes(cls, changes):
        class Serializer(json.JSONEncoder):
            def default(self, o):
                if isinstance(o, datetime.datetime):
                    return o.strftime("%Y-%m-%d %H:%M:%S")
                elif isinstance(o, datetime.date):
                    return o.strftime("%Y-%m-%d")
                else:
                    return json.JSONEncoder.default(self, o)
        return json.dumps(changes, cls=Serializer)

    def get_changes_list(self):
        return sorted((k, v[0], v[1]) for k, v in list(json.loads(self.changes).items()))