VRE Backend API and Scheduler
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 

273 lines
12 KiB

from django.conf import settings
from django.contrib.staticfiles import finders
from django.db import models
from django.db.models import Count, Sum
from django.template.loader import render_to_string
from django.urls import reverse
from django.utils import timezone
from django.utils.translation import gettext_lazy as _
from rest_framework.exceptions import ValidationError
from lib.models.base import MetaDataModel
from lib.utils.emails import EmailMultiRelated
from lib.utils.general import file_upload_to, get_random_int_value, remove_html_tags
from apps.researcher.models import Researcher
from apps.university.models import StudyField
from pathlib import Path
import uuid
import jwt
import datetime
class StudyManager(models.Manager):
"""
This is a custom study manager for getting extra information from a study model like:
- Total files / datadrops attached for a study.
- Total file size of all the data drops for a study.
- Total invitations that have been sent for a study.
"""
def get_queryset(self):
"""
Returns the queryset with extra fields
- '_total_files',
- '_total_file_size'
- '_total_invitations'.
This will overrule/alter the existing queryset.
"""
return super().get_queryset().select_related('storagelocation').annotate(
_total_files=Count('files', distinct=True),
_total_file_size=Sum('files__filesize', distinct=True),
_total_invitations=Count('invitations', distinct=True),
)
class StudyRoleNames(models.TextChoices):
"""Research study role options
Valid status values:
- Administrator
- Researcher
- Member
"""
ADMIN = ('ADMIN', _('Administrator'))
CONTRIBUTOR = ('CONTRIBUTOR', _('Contributor'))
class Study(MetaDataModel):
"""
A model to represent a study of a researcher. This study will receive the dropoff files from various sources.
It will inherit the attributes :attr:`~lib.models.base.MetaDataModel.created_at` and :attr:`~lib.models.base.MetaDataModel.updated_at` from the Abstract model :class:`~lib.models.base.MetaDataModel`
Attributes
----------
researcher : Researcher
The Django Researcher model that is the owner of this study.
name : str
The name of the study. Can be entered freely by the researcher.
description : str
A small description what the study is about. This will be used on the upload page and invitation mail.
upload_code : str
A unique code that is used as a token for uploading. This is a general upload code. Every Invitation will get his own upload code as well. This will be auto generated.
upload_uuid : uuid4
A UUID v4 string that is used to create a unique upload url. This will be auto generated when created.
"""
class Meta:
verbose_name = _('studie')
verbose_name_plural = _('studies')
ordering = ['name']
owner = models.ForeignKey(Researcher, verbose_name=Researcher._meta.verbose_name, on_delete=models.CASCADE, help_text=_('The researcher that is the owner this study. By default the researcher that has created this study.'))
name = models.CharField(_('Name'), max_length=200, help_text=_('Name of the research study.'))
description = models.TextField(_('Description'), blank=True, null=True, help_text=_('Enter a short description for this study.'))
avatar = models.ImageField(_('Avatar'), upload_to=file_upload_to, blank=True, null=True)
code = models.CharField(_('Code'), max_length=50, help_text=_('The research study code.'))
human_subject = models.BooleanField(_('Human subject'), help_text=_('Is this research study using real humans.'))
field = models.ForeignKey(StudyField, verbose_name=StudyField._meta.verbose_name, on_delete=models.CASCADE, help_text=_('The study field for this reaserch study.'))
contributors = models.ManyToManyField(Researcher, related_name='contributors', through='StudyRole', through_fields=('study', 'researcher'))
# These are used for accepting files (datadrops)
upload_code = models.CharField(_('Upload code'), max_length=20, default=get_random_int_value, editable=False, help_text=_('A unique upload code. Will be generated when a new study is saved.'))
upload_uuid = models.UUIDField(_('Upload url key'), unique=True, default=uuid.uuid4, editable=False, help_text=_('A unique upload url. Will be generated when a new study is saved.'))
# Here we load our custom StudyManager so we always have the amount of files, total file size and amount of invitations directly accessable
objects = StudyManager()
def is_an_admin(self, researcher):
return self.contributors.filter(studyrole__researcher=researcher, studyrole__role=StudyRoleNames.ADMIN).count() == 1
@property
def has_storage(self):
"""boolean: Returns true when there is at least one storage location connected to this study"""
# print(dir(self.storagelocation))
return self.storagelocation is not None
@property
def get_absolute_url(self):
"""str: Returns the full url to the study detail page."""
return 'aaa'
# return reverse('study:detail', kwargs={'study_id': self.pk})
@property
def web_upload_url(self):
"""str: Returns the full url for the study upload page through web."""
return f'{settings.DROPOFF_BASE_URL.strip("/")}/{self.upload_uuid}'
@property
def api_upload_url(self):
"""str: Returns the full url for the study upload API entrypoint."""
return f'{settings.DROPOFF_UPLOAD_HOST.strip("/")}/files/{self.upload_uuid}/'
@property
def total_files(self):
"""int: Returns the total amount of uploaded files"""
return self._total_files
@property
def total_file_size(self):
"""int: Returns the total upload amount of the uploaded files"""
return 0 if self._total_file_size is None else self._total_file_size
@property
def total_invitations(self):
"""int: Returns the total amount of invitations"""
return self._total_invitations
def __str__(self):
"""str: Returns a readable string for the study."""
return f'{self.name}'
class StudyRole(MetaDataModel):
study = models.ForeignKey(Study, verbose_name=Study._meta.verbose_name, on_delete=models.CASCADE, help_text=_('Study'))
researcher = models.ForeignKey(Researcher, verbose_name=Researcher._meta.verbose_name, on_delete=models.CASCADE, help_text=_('Researcher'))
role = models.CharField(_('Role'), max_length=15, choices=StudyRoleNames.choices, default=StudyRoleNames.CONTRIBUTOR, help_text=_('The role withing this research study.'))
# By default active is False. Will be set to true if the invitation is valid and the researcher becomes an active member of this study
active = models.BooleanField(_('Active'), default=False, help_text=_('Is the researcher an active member of the study'))
invited_at = models.DateTimeField(_('Date invited'), default=timezone.now, help_text=_('The date and time when the invitation has been sent'))
class Meta:
constraints = [
models.UniqueConstraint(fields=['study', 'researcher'], name='unique_researcher_study_role'),
]
def sent_invitation_email(self, sender=None):
# Generate JWT token
jwt_data = {
'timestamp': int(self.invited_at.timestamp()),
'study': self.study.pk,
'contributor': self.pk,
'researcher': self.researcher.pk,
'exp': datetime.datetime.now(tz=timezone.utc) + datetime.timedelta(seconds=settings.STUDY_INVITATION_LINK_EXPIRE_DURATION)
}
jwt_token = jwt.encode(payload=jwt_data, key=settings.SECRET_KEY, algorithm="HS256")
url = f'{settings.STUDY_INVITATION_LINK_DOMAIN}' + reverse('api:v1:study-invite-join', kwargs={'study_id': self.study.pk, 'jwt_token': jwt_token})
template_variables = {
'researcher': self.researcher.display_name,
'study': self.study.name,
'invitation_link': url
}
from_email = settings.EMAIL_FROM_ADDRESS
subject = render_to_string('email/invitation_to_study/subject.txt', template_variables)
html_message = render_to_string('email/invitation_to_study/body.html', template_variables)
text_message = render_to_string('email/invitation_to_study/body.txt', template_variables)
if '' == text_message:
text_message = remove_html_tags(html_message)
reply_to_headers = {
'reply-to': f'{self.researcher.display_name}<{self.researcher.user.email}>'
}
if sender is not None:
reply_to_headers['reply-to'] = f'{sender.display_name}<{sender.user.email}>'
message = EmailMultiRelated(subject,
text_message,
from_email,
[f'{self.researcher.display_name}<{self.researcher.user.email}>'],
headers=reply_to_headers)
message.attach_alternative(html_message, 'text/html')
logo = Path(finders.find('images/RUG_Logo.jpg'))
if logo.is_file():
message.attach_related_file(logo)
message_sent = message.send() == 1
if message_sent:
self.invited_at = timezone.now()
self.save()
return message_sent
def check_invitation_link(self, researcher, jwt_token):
jwt_data = None
try:
jwt_data = jwt.decode(jwt_token, settings.SECRET_KEY, algorithms=["HS256"])
except jwt.ExpiredSignatureError:
raise ValidationError(_('Invitation token is expired. Please ask for a new invitation'))
if jwt_data is None:
return False
if jwt_data['researcher'] == researcher.pk and jwt_data['study'] == self.study.pk:
return True
else:
raise ValidationError(_('Unfortunately the token contains invalid data'))
def send_remove_notification_email(self, sender=None):
template_variables = {
'researcher': self.researcher.display_name,
'study': self.study.name,
'sender': self.study.owner.display_name if sender is None else sender.display_name
}
from_email = settings.EMAIL_FROM_ADDRESS
subject = render_to_string('email/removed_from_study/subject.txt', template_variables)
html_message = render_to_string('email/removed_from_study/body.html', template_variables)
text_message = render_to_string('email/removed_from_study/body.txt', template_variables)
if '' == text_message:
text_message = remove_html_tags(html_message)
reply_to_headers = {
'reply-to': f'{self.researcher.display_name}<{self.researcher.user.email}>'
}
if sender is not None:
reply_to_headers['reply-to'] = f'{sender.display_name}<{sender.user.email}>'
message = EmailMultiRelated(subject,
text_message,
from_email,
[f'{self.researcher.display_name}<{self.researcher.user.email}>'],
headers=reply_to_headers)
message.attach_alternative(html_message, 'text/html')
logo = Path(finders.find('images/RUG_Logo.jpg'))
if logo.is_file():
message.attach_related_file(logo)
return message.send() == 1
def __str__(self):
"""str: Returns a readable string for the study role."""
return f'{self.researcher.display_name} is {self.role} of {self.study.name}'