Skip to content
Snippets Groups Projects
Commit 325d707e authored by Carlos Vega's avatar Carlos Vega
Browse files

use decorator in signals

parent 69ef20c0
No related branches found
No related tags found
2 merge requests!462Redcap/generic test instance,!461Db/mariadb
Hide whitespace changes
Inline Side-by-side
smash/web/auth.py
+ 6
− 7
View file @ 325d707e
......@@ -8,9 +8,8 @@ logger = logging.getLogger(__name__)
def do_login(request):
user_login = request.POST.get('username', 'none')
user = authenticate(username=user_login,
password=request.POST.get('password', 'none'))
user_login = request.POST.get("username", "none")
user = authenticate(username=user_login, password=request.POST.get("password", "none"))
if user is not None:
login(request, user)
return True, "ok"
......@@ -31,15 +30,15 @@ def do_logout(request):
def user_logged_in_callback(sender, request, user, **kwargs): # pylint: disable=unused-argument
# to cover more complex cases:
# http://stackoverflow.com/questions/4581789/how-do-i-get-user-ip-address-in-django
ip = request.META.get('REMOTE_ADDR')
logger.info('login user: %s via ip: %s', user, ip)
ip = request.META.get("REMOTE_ADDR")
logger.info("login user: %s via ip: %s", user, ip)
@receiver(user_logged_out)
def user_logged_out_callback(sender, request, user, **kwargs): # pylint: disable=unused-argument
ip = request.META.get('REMOTE_ADDR')
ip = request.META.get("REMOTE_ADDR")
logger.info('logout user: %s via ip: %s', user, ip)
logger.info("logout user: %s via ip: %s", user, ip)
@receiver(user_login_failed)
......
smash/web/models/privacy_notice.py
+ 12
− 8
View file @ 325d707e
......@@ -5,19 +5,20 @@ from django.db import models
from django.dispatch import receiver
from web.templatetags.filters import basename
from web import disable_for_loaddata
class PrivacyNotice(models.Model):
name = models.CharField(max_length=255, verbose_name='Name')
created_at = models.DateTimeField(auto_now_add=True, verbose_name='Created at')
updated_at = models.DateTimeField(auto_now=True, verbose_name='Updated at')
summary = models.CharField(max_length=255, verbose_name='Summary', blank=False, null=False)
document = models.FileField(upload_to='privacy_notices/',
verbose_name='Study Privacy Notice file',
null=False, editable=True)
name = models.CharField(max_length=255, verbose_name="Name")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="Created at")
updated_at = models.DateTimeField(auto_now=True, verbose_name="Updated at")
summary = models.CharField(max_length=255, verbose_name="Summary", blank=False, null=False)
document = models.FileField(
upload_to="privacy_notices/", verbose_name="Study Privacy Notice file", null=False, editable=True
)
def __str__(self):
return f'{self.name} ({basename(self.document.url)})'
return f"{self.name} ({basename(self.document.url)})"
@property
def all_studies(self):
......@@ -26,7 +27,9 @@ class PrivacyNotice(models.Model):
# These two auto-delete files from filesystem when they are unneeded:
@receiver(models.signals.post_delete, sender=PrivacyNotice)
@disable_for_loaddata
def auto_delete_file_on_delete(sender, instance: PrivacyNotice, **kwargs): # pylint: disable=unused-argument
"""
Deletes file from filesystem
......@@ -38,6 +41,7 @@ def auto_delete_file_on_delete(sender, instance: PrivacyNotice, **kwargs): # py
@receiver(models.signals.pre_save, sender=PrivacyNotice)
@disable_for_loaddata
def auto_delete_file_on_change(sender, instance: PrivacyNotice, **kwargs): # pylint: disable=unused-argument
"""
Deletes old file from filesystem
......
smash/web/models/study_subject.py
+ 7
− 9
View file @ 325d707e
......@@ -10,6 +10,7 @@ from django.dispatch import receiver
from web.models import Appointment, Location, Provenance, Visit, VoucherType
from web.models.constants import BOOL_CHOICES, FILE_STORAGE
from web.models.custom_data import CustomStudySubjectField, CustomStudySubjectValue
from web import disable_for_loaddata
logger = logging.getLogger(__name__)
......@@ -214,16 +215,12 @@ class StudySubject(models.Model):
def custom_data_values(self):
# find the custom fields that have not yet been populated into the study subject
# https://docs.djangoproject.com/en/3.2/ref/models/querysets/#filteredrelation-objects
fields = (
CustomStudySubjectField.objects
.annotate(
t=FilteredRelation(
"customstudysubjectvalue",
condition=Q(customstudysubjectvalue__study_subject=self),
)
fields = CustomStudySubjectField.objects.annotate(
t=FilteredRelation(
"customstudysubjectvalue",
condition=Q(customstudysubjectvalue__study_subject=self),
)
.filter(t__study_subject_field__isnull=True, study=self.study)
)
).filter(t__study_subject_field__isnull=True, study=self.study)
for field in fields:
CustomStudySubjectValue.objects.create(
......@@ -273,6 +270,7 @@ class StudySubject(models.Model):
# SIGNALS
@receiver(post_save, sender=StudySubject)
@disable_for_loaddata
def set_as_resigned_or_excluded_or_endpoint_reached(sender, instance, **kwargs): # pylint: disable=unused-argument
if instance.excluded:
instance.mark_as_excluded()
......
smash/web/models/subject.py
+ 61
− 107
View file @ 325d707e
......@@ -7,6 +7,7 @@ from django.dispatch import receiver
from .constants import SEX_CHOICES, COUNTRY_OTHER_ID
from web.models import Country, Visit, Appointment, Provenance
from web import disable_for_loaddata
from . import Language
logger = logging.getLogger(__name__)
......@@ -14,121 +15,69 @@ logger = logging.getLogger(__name__)
class Subject(models.Model):
class Meta:
app_label = 'web'
app_label = "web"
permissions = [
("export_subjects", "Can export subject data to excel/csv"),
]
@property
def provenances(self):
return Provenance.objects.filter(modified_table=Subject._meta.db_table, modified_table_id=self.id)\
.order_by('-modification_date')
sex = models.CharField(max_length=1,
choices=SEX_CHOICES,
verbose_name='Sex'
)
first_name = models.CharField(max_length=50,
verbose_name='First name'
)
social_security_number = models.CharField(max_length=50,
verbose_name='Social security number',
blank=True,
)
last_name = models.CharField(max_length=50,
verbose_name='Last name'
)
languages = models.ManyToManyField(Language,
blank=True,
verbose_name='Known languages'
)
default_written_communication_language = models.ForeignKey(Language,
null=True,
blank=True,
related_name="subjects_written_communication",
verbose_name='Default language for document generation',
on_delete=models.SET_NULL
)
phone_number = models.CharField(max_length=64,
null=True,
blank=True,
verbose_name='Phone number'
)
phone_number_2 = models.CharField(max_length=64,
null=True,
blank=True,
verbose_name='Phone number 2'
)
phone_number_3 = models.CharField(max_length=64,
null=True,
blank=True,
verbose_name='Phone number 3'
)
email = models.EmailField(
null=True,
return Provenance.objects.filter(modified_table=Subject._meta.db_table, modified_table_id=self.id).order_by(
"-modification_date"
)
sex = models.CharField(max_length=1, choices=SEX_CHOICES, verbose_name="Sex")
first_name = models.CharField(max_length=50, verbose_name="First name")
social_security_number = models.CharField(
max_length=50,
verbose_name="Social security number",
blank=True,
verbose_name='E-mail'
)
date_born = models.DateField(
last_name = models.CharField(max_length=50, verbose_name="Last name")
languages = models.ManyToManyField(Language, blank=True, verbose_name="Known languages")
default_written_communication_language = models.ForeignKey(
Language,
null=True,
blank=True,
verbose_name='Date of birth (YYYY-MM-DD)'
related_name="subjects_written_communication",
verbose_name="Default language for document generation",
on_delete=models.SET_NULL,
)
phone_number = models.CharField(max_length=64, null=True, blank=True, verbose_name="Phone number")
phone_number_2 = models.CharField(max_length=64, null=True, blank=True, verbose_name="Phone number 2")
phone_number_3 = models.CharField(max_length=64, null=True, blank=True, verbose_name="Phone number 3")
email = models.EmailField(null=True, blank=True, verbose_name="E-mail")
address = models.CharField(max_length=255,
blank=True,
verbose_name='Address'
)
postal_code = models.CharField(max_length=7,
blank=True,
verbose_name='Postal code'
)
city = models.CharField(max_length=50,
blank=True,
verbose_name='City'
)
country = models.ForeignKey(Country,
null=False,
blank=False,
default=COUNTRY_OTHER_ID,
verbose_name='Country', on_delete=models.CASCADE
)
next_of_kin_name = models.CharField(max_length=255,
blank=True,
verbose_name='Next of kin'
)
next_of_kin_phone = models.CharField(max_length=50,
blank=True,
verbose_name='Next of kin phone'
)
next_of_kin_address = models.TextField(max_length=2000,
blank=True,
verbose_name='Next of kin address'
)
dead = models.BooleanField(
verbose_name='Deceased',
default=False,
editable=True
date_born = models.DateField(null=True, blank=True, verbose_name="Date of birth (YYYY-MM-DD)")
address = models.CharField(max_length=255, blank=True, verbose_name="Address")
postal_code = models.CharField(max_length=7, blank=True, verbose_name="Postal code")
city = models.CharField(max_length=50, blank=True, verbose_name="City")
country = models.ForeignKey(
Country, null=False, blank=False, default=COUNTRY_OTHER_ID, verbose_name="Country", on_delete=models.CASCADE
)
next_of_kin_name = models.CharField(max_length=255, blank=True, verbose_name="Next of kin")
next_of_kin_phone = models.CharField(max_length=50, blank=True, verbose_name="Next of kin phone")
next_of_kin_address = models.TextField(max_length=2000, blank=True, verbose_name="Next of kin address")
dead = models.BooleanField(verbose_name="Deceased", default=False, editable=True)
def pretty_address(self):
return f'{self.address} ({self.postal_code}), {self.city}. {self.country}'
return f"{self.address} ({self.postal_code}), {self.city}. {self.country}"
def mark_as_dead(self):
self.dead = True
......@@ -142,8 +91,9 @@ class Subject(models.Model):
visit.save()
def finish_all_appointments(self):
appointments = Appointment.objects.filter(visit__subject__subject=self,
status=Appointment.APPOINTMENT_STATUS_SCHEDULED)
appointments = Appointment.objects.filter(
visit__subject__subject=self, status=Appointment.APPOINTMENT_STATUS_SCHEDULED
)
for appointment in appointments:
appointment.status = Appointment.APPOINTMENT_STATUS_CANCELLED
appointment.save()
......@@ -154,13 +104,17 @@ class Subject(models.Model):
# SIGNALS
@receiver(post_save, sender=Subject)
@disable_for_loaddata
def set_as_deceased(sender, instance, **kwargs): # pylint: disable=unused-argument
if instance.dead:
p = Provenance(modified_table=Subject._meta.db_table,
modified_table_id=instance.id, modification_author=None,
previous_value=instance.dead, new_value=True,
modification_description=f'Subject "{instance}" marked as dead',
modified_field='dead',
)
p = Provenance(
modified_table=Subject._meta.db_table,
modified_table_id=instance.id,
modification_author=None,
previous_value=instance.dead,
new_value=True,
modification_description=f'Subject "{instance}" marked as dead',
modified_field="dead",
)
instance.mark_as_dead()
p.save()
smash/web/models/visit.py
+ 59
− 48
View file @ 325d707e
......@@ -8,61 +8,59 @@ from django.db.models.signals import post_save
from django.dispatch import receiver
from web.models.constants import BOOL_CHOICES
from web import disable_for_loaddata
logger = logging.getLogger(__name__)
class Visit(models.Model):
class Meta:
app_label = 'web'
app_label = "web"
subject = models.ForeignKey("web.StudySubject", on_delete=models.CASCADE, verbose_name='Subject'
)
datetime_begin = models.DateTimeField(
verbose_name='Visit starts at'
)
subject = models.ForeignKey("web.StudySubject", on_delete=models.CASCADE, verbose_name="Subject")
datetime_begin = models.DateTimeField(verbose_name="Visit starts at")
datetime_end = models.DateTimeField(
verbose_name='Visit ends at'
verbose_name="Visit ends at"
) # Deadline before which all appointments need to be scheduled
is_finished = models.BooleanField(
verbose_name='Has ended',
default=False
is_finished = models.BooleanField(verbose_name="Has ended", default=False)
post_mail_sent = models.BooleanField(choices=BOOL_CHOICES, verbose_name="Post mail sent", default=False)
appointment_types = models.ManyToManyField(
"web.AppointmentType",
verbose_name="Requested appointments",
blank=True,
)
post_mail_sent = models.BooleanField(choices=BOOL_CHOICES,
verbose_name='Post mail sent',
default=False
)
appointment_types = models.ManyToManyField("web.AppointmentType",
verbose_name='Requested appointments',
blank=True,
)
# this value is automatically computed by signal handled by
# update_visit_number method
visit_number = models.IntegerField(
verbose_name='Visit number',
default=1
)
visit_number = models.IntegerField(verbose_name="Visit number", default=1)
@property
def next_visit(self):
return Visit.objects.filter(subject=self.subject, visit_number=self.visit_number + 1) \
.order_by('datetime_begin', 'datetime_end').first()
return (
Visit.objects.filter(subject=self.subject, visit_number=self.visit_number + 1)
.order_by("datetime_begin", "datetime_end")
.first()
)
@property
def future_visits(self):
return Visit.objects.filter(subject=self.subject).filter(visit_number__gt=self.visit_number) \
.order_by('datetime_begin', 'datetime_end')
return (
Visit.objects.filter(subject=self.subject)
.filter(visit_number__gt=self.visit_number)
.order_by("datetime_begin", "datetime_end")
)
def __str__(self):
start = self.datetime_begin.strftime('%Y-%m-%d')
end = self.datetime_end.strftime('%Y-%m-%d')
finished = '✓' if self.is_finished else ''
return f'#{self.visit_number:02} ' \
f'| {start} / {end} ' \
f'| {self.subject.subject.first_name} {self.subject.subject.last_name} ' \
f'| {finished}'
start = self.datetime_begin.strftime("%Y-%m-%d")
end = self.datetime_end.strftime("%Y-%m-%d")
finished = "✓" if self.is_finished else ""
return (
f"#{self.visit_number:02} "
f"| {start} / {end} "
f"| {self.subject.subject.first_name} {self.subject.subject.last_name} "
f"| {finished}"
)
def mark_as_finished(self):
self.is_finished = True
......@@ -96,32 +94,36 @@ class Visit(models.Model):
time_to_next_visit = relativedelta(**args) * (follow_up_number - start_number)
logger.warning('new visit: %s %s %s', args, relativedelta(**args), time_to_next_visit)
logger.warning("new visit: %s %s %s", args, relativedelta(**args), time_to_next_visit)
Visit.objects.create(
subject=self.subject,
datetime_begin=visit_started + time_to_next_visit,
datetime_end=visit_started + time_to_next_visit + relativedelta(
months=study.default_visit_duration_in_months)
datetime_end=visit_started
+ time_to_next_visit
+ relativedelta(months=study.default_visit_duration_in_months),
)
def unfinish(self):
# if ValueError messages are changed, change test/view/test_visit.py
# check visit is indeed finished
if not self.is_finished:
raise ValueError('The visit is not finished.')
raise ValueError("The visit is not finished.")
# check if there are some unfinished visits before this visit
unfinished_visits = Visit.objects.filter(subject=self.subject,
is_finished=False, datetime_begin__lt=self.datetime_begin).count()
unfinished_visits = Visit.objects.filter(
subject=self.subject, is_finished=False, datetime_begin__lt=self.datetime_begin
).count()
if unfinished_visits > 0:
raise ValueError("Visit can't be unfinished. There is at least one unfinished visit.")
# check that there is only one future visit
future_visits = self.future_visits
if len(future_visits) > 1:
raise ValueError("Visit can't be unfinished. "
"Only visits with one immediate future visit (without appointments) can be unfinished.")
raise ValueError(
"Visit can't be unfinished. "
"Only visits with one immediate future visit (without appointments) can be unfinished."
)
elif len(future_visits) == 1:
# check that the future visit has no appointments
# remove visit if it has no appointments
......@@ -143,6 +145,7 @@ class Visit(models.Model):
@receiver(post_save, sender=Visit)
@disable_for_loaddata
def check_visit_number(sender, instance, created, **kwargs): # pylint: disable=unused-argument
# no other solution to ensure the visit_number is in chronological order than to sort the whole list if there are
# future visits
......@@ -150,14 +153,22 @@ def check_visit_number(sender, instance, created, **kwargs): # pylint: disable=
if visit.subject is not None: # not sure if select_for_update has an effect, the tests work as well without it
# new visit, sort only future visit respect to the new one
if created:
visits_before = Visit.objects.select_for_update().filter(subject=visit.subject) \
.filter(datetime_begin__lt=visit.datetime_begin).count()
visits_before = (
Visit.objects.select_for_update()
.filter(subject=visit.subject)
.filter(datetime_begin__lt=visit.datetime_begin)
.count()
)
# we need to sort the future visits respect to the new one, if any
visits = Visit.objects.select_for_update().filter(subject=visit.subject) \
.filter(datetime_begin__gte=visit.datetime_begin).order_by('datetime_begin', 'id')
visits = (
Visit.objects.select_for_update()
.filter(subject=visit.subject)
.filter(datetime_begin__gte=visit.datetime_begin)
.order_by("datetime_begin", "id")
)
with transaction.atomic(): # not sure if it has an effect, the tests work as well without it
for i, v in enumerate(visits):
expected_visit_number = (visits_before + i + 1)
expected_visit_number = visits_before + i + 1
if v.visit_number != expected_visit_number:
# does not rise post_save, we avoid recursion
Visit.objects.filter(id=v.id).update(visit_number=expected_visit_number)
......@@ -167,9 +178,9 @@ def check_visit_number(sender, instance, created, **kwargs): # pylint: disable=
visit.visit_number = v.visit_number
else:
# if visits are modified, then, check everything
visits = Visit.objects.select_for_update().filter(subject=visit.subject).order_by('datetime_begin', 'id')
visits = Visit.objects.select_for_update().filter(subject=visit.subject).order_by("datetime_begin", "id")
with transaction.atomic():
for i, v in enumerate(visits):
expected_visit_number = (i + 1)
expected_visit_number = i + 1
if v.visit_number != expected_visit_number: # update only those with wrong numbers
Visit.objects.filter(id=v.id).update(visit_number=expected_visit_number)
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment