"""Models for the timeline app"""
import logging
import uuid
from datetime import datetime
from typing import Any, Optional, Union
from django.conf import settings
from django.db import models
from django.db.models import Max, Q, QuerySet
# Projectroles dependency
from projectroles.models import Project
logger = logging.getLogger(__name__)
# Access Django user model
AUTH_USER_MODEL = getattr(settings, 'AUTH_USER_MODEL', 'auth.User')
# Local constants
TL_STATUS_OK = 'OK'
TL_STATUS_INIT = 'INIT'
TL_STATUS_SUBMIT = 'SUBMIT'
TL_STATUS_FAILED = 'FAILED'
TL_STATUS_INFO = 'INFO'
TL_STATUS_CANCEL = 'CANCEL'
EVENT_STATUS_TYPES = [
TL_STATUS_OK,
TL_STATUS_INIT,
TL_STATUS_SUBMIT,
TL_STATUS_FAILED,
TL_STATUS_INFO,
TL_STATUS_CANCEL,
]
DEFAULT_MESSAGES = {
TL_STATUS_OK: 'All OK',
TL_STATUS_INIT: 'Event initialized',
TL_STATUS_SUBMIT: 'Job submitted asynchronously',
TL_STATUS_FAILED: 'Failed (unknown problem)',
TL_STATUS_INFO: 'Info level action',
TL_STATUS_CANCEL: 'Action cancelled',
}
OBJ_REF_UNNAMED = '(unnamed)'
[docs]
class TimelineEventManager(models.Manager):
"""Manager for custom table-level TimelineEvent queries"""
[docs]
def get_object_events(
self,
project: Optional[Project],
object_model: str,
object_uuid: Union[str, uuid.UUID],
order_by: str = '-pk',
) -> QuerySet:
"""
Return events which are linked to an object reference.
:param project: Project object or None
:param object_model: Object model (string)
:param object_uuid: sodar_uuid of the original object
:param order_by: Ordering (default = pk descending)
:return: QuerySet
"""
return TimelineEvent.objects.filter(
project=project,
event_objects__object_model=object_model,
event_objects__object_uuid=object_uuid,
).order_by(order_by)
[docs]
def find(
self,
search_terms: list[str],
projects: QuerySet[Project],
keywords: Optional[dict] = None,
) -> QuerySet:
"""
Return events matching the query.
:param search_terms: Search terms (list of strings)
:param projects: QuerySet of projects where the terms are searched
:param keywords: Optional search keywords as key/value pairs (dict)
:return: QuerySet of TimelineEvent objects
"""
search_limit = getattr(settings, 'TIMELINE_SEARCH_LIMIT', 250)
objects = (
super()
.get_queryset()
.filter(Q(project__in=projects) | Q(project__isnull=True))
)
term_query = Q()
for t in search_terms:
term_query.add(Q(event_name__icontains=t), Q.OR)
term_query.add(Q(event_name__icontains=t.replace(' ', '_')), Q.OR)
term_query.add(Q(description__icontains=t), Q.OR)
term_query.add(Q(event_objects__name__icontains=t), Q.OR)
term_query.add(Q(user__name__icontains=t), Q.OR)
term_query.add(Q(user__username__icontains=t), Q.OR)
items = (
objects.filter(term_query)
.annotate(timestamp=Max('status_changes__timestamp'))
.order_by('-timestamp')
)
return items[:search_limit]
[docs]
class TimelineEvent(models.Model):
"""
Class representing a Project event. Can also be a site-wide event not linked
to a specific project.
"""
#: Project to which the event belongs
project = models.ForeignKey(
Project,
related_name='events',
help_text='Project to which the event belongs (null for no project)',
on_delete=models.CASCADE,
null=True,
)
#: App from which the event was triggered
app = models.CharField(
max_length=255, help_text='App from which the event was triggered'
)
#: Plugin to which the event is related (optional)
plugin = models.CharField(
max_length=255,
blank=True,
null=True,
help_text='Plugin to which the event is related (optional, if unset '
'plugin with the name of the app is assumed)',
)
#: User who initiated the event (optional)
user = models.ForeignKey(
AUTH_USER_MODEL,
null=True,
help_text='User who initiated the event (optional)',
on_delete=models.CASCADE,
)
#: Event ID string
event_name = models.CharField(max_length=255, help_text='Event ID string')
#: Description of status change (may include {object_name} references)
description = models.TextField(
help_text='Description of status change '
'(may include {object label} references)'
)
#: Additional event data as JSON
extra_data = models.JSONField(
default=dict, help_text='Additional event data as JSON'
)
#: Event is classified (only viewable by user levels specified in rules)
classified = models.BooleanField(
default=False,
help_text='Event is classified (only viewable by user levels '
'specified in rules)',
)
#: UUID for the event
sodar_uuid = models.UUIDField(
default=uuid.uuid4, unique=True, help_text='Event SODAR UUID'
)
# Set manager for custom queries
objects = TimelineEventManager()
def __str__(self):
return '{}{}{}'.format(
(self.project.title + ': ') if self.project else '',
self.event_name,
('/' + self.user.username) if self.user else '',
)
def __repr__(self):
return 'TimelineEvent({})'.format(
', '.join(repr(v) for v in self.get_repr_values())
)
def get_repr_values(self) -> list[str]:
return [
self.project.title if self.project else 'N/A',
self.event_name,
self.user.username if self.user else 'N/A',
]
[docs]
def get_status(self) -> Optional['TimelineEventStatus']:
"""Return the current event status"""
return self.status_changes.order_by('-timestamp').first()
[docs]
def get_timestamp(self) -> datetime:
"""Return the timestamp of current status"""
return self.status_changes.order_by('-timestamp').first().timestamp
[docs]
def get_status_changes(self, reverse: bool = False) -> QuerySet:
"""Return all status changes for the event"""
return self.status_changes.order_by(
'{}pk'.format('-' if reverse else '')
)
[docs]
def get_project(self) -> Project:
"""Return the project for the event"""
return self.project
[docs]
def add_object(
self, obj: Any, label: str, name: str, extra_data: Optional[dict] = None
) -> 'TimelineEventObjectRef':
"""
Add object reference to an event.
:param obj: Django object to which we want to refer
:param label: Label for the object in the event description (string)
:param name: Name or title of the object (string)
:param extra_data: Additional data related to object (dict, optional)
:return: TimelineEventObjectRef object
"""
if not name:
logger.warning(
'Adding object reference with no name: "{}" ({})'.format(
obj, getattr(obj, 'sodar_uuid')
)
)
name = OBJ_REF_UNNAMED
ref = TimelineEventObjectRef(
event=self,
label=label,
name=name,
object_model=obj.__class__.__name__,
object_uuid=obj.sodar_uuid,
extra_data=extra_data or {},
)
ref.save()
if settings.DEBUG:
logger.debug(
f'Add timeline object ref for {ref.object_model} "{name}" '
f'(object={str(obj.sodar_uuid)}; '
f'event={self.sodar_uuid}; '
f'uuid={ref.sodar_uuid})'
)
return ref
[docs]
def set_status(
self,
status_type: str,
status_desc: Optional[str] = None,
extra_data: Optional[dict] = None,
) -> 'TimelineEventStatus':
"""
Set event status.
:param status_type: Status type string (see EVENT_STATUS_TYPES)
:param status_desc: Description string (optional)
:param extra_data: Extra data for the status (dict, optional)
:return: TimelineEventStatus object
:raise: TypeError if status_type is invalid
"""
if status_type not in EVENT_STATUS_TYPES:
raise TypeError(
'Invalid status type (accepted values: {})'.format(
', '.join(v for v in EVENT_STATUS_TYPES)
)
)
status = TimelineEventStatus(
event=self,
status_type=status_type,
description=(
status_desc if status_desc else DEFAULT_MESSAGES[status_type]
),
extra_data=extra_data or {},
)
status.save()
if settings.DEBUG and status_type != TL_STATUS_INIT:
logger.debug(
f'Set timeline event {self.app}.{self.event_name} '
f'status to {status_type} '
f'(event={self.sodar_uuid}; '
f'uuid={status.sodar_uuid})'
)
return status
[docs]
class TimelineEventObjectRef(models.Model):
"""
Class representing a reference to an object (existing or removed)
related to a timeline event status.
"""
#: Event to which the object belongs
event = models.ForeignKey(
TimelineEvent,
related_name='event_objects',
help_text='Event to which the object belongs',
on_delete=models.CASCADE,
)
#: Label for the object related to the event
label = models.CharField(
max_length=255,
null=False,
blank=False,
help_text='Label for the object related to the event',
)
#: Name or title of the object
name = models.CharField(
max_length=255,
null=False,
blank=False,
help_text='Name or title of the object',
)
#: Object model as string
object_model = models.CharField(
max_length=255,
null=False,
blank=False,
help_text='Object model as string',
)
#: Object SODAR UUID
object_uuid = models.UUIDField(
null=True, blank=True, unique=False, help_text='Object SODAR UUID'
)
#: Additional data related to the object as JSON
extra_data = models.JSONField(
default=dict, help_text='Additional data related to the object as JSON'
)
#: UUID for this object reference
sodar_uuid = models.UUIDField(
default=uuid.uuid4, unique=True, help_text='Object reference SODAR UUID'
)
def __str__(self):
return '{} ({})'.format(
self.event.__str__(),
self.name,
)
def __repr__(self):
values = self.event.get_repr_values() + [self.name]
return 'TimelineEventObjectRef({})'.format(
', '.join(repr(v) for v in values)
)
[docs]
def get_project(self) -> Project:
"""Return the project for the event"""
return self.event.project
[docs]
class TimelineEventStatus(models.Model):
"""Class representing a timeline event status"""
#: Event to which the status change belongs
event = models.ForeignKey(
TimelineEvent,
related_name='status_changes',
help_text='Event to which the status change belongs',
on_delete=models.CASCADE,
)
#: DateTime of the status change
timestamp = models.DateTimeField(
auto_now_add=True, help_text='DateTime of the status change'
)
#: Type of the status change
status_type = models.CharField(
max_length=64,
null=False,
blank=False,
help_text='Type of the status change',
)
#: Description of status change (optional)
description = models.TextField(
blank=True, help_text='Description of status change (optional)'
)
#: Additional status data as JSON
extra_data = models.JSONField(
default=dict, help_text='Additional status data as JSON'
)
#: UUID for the status
sodar_uuid = models.UUIDField(
default=uuid.uuid4, unique=True, help_text='Status SODAR UUID'
)
def __str__(self):
return '{} ({})'.format(
self.event.__str__(),
self.status_type,
)
def __repr__(self):
values = self.event.get_repr_values() + [self.status_type]
return 'TimelineEventStatus({})'.format(
', '.join(repr(v) for v in values)
)
[docs]
def get_project(self) -> Project:
"""Return the project for the event"""
return self.event.project