Submission
pretalx.submission.models.submission.Submission source
Submissions are, next to Event, the central model in pretalx.
State changes must go through pretalx.submission.domain.submission.set_submission_state(), which is called by the accept(), reject() etc model methods.
Fields 27
idAutoFieldpkcodeCharFielduniquesource
code = models.CharField(max_length=16, unique=True)
eventForeignKeysource
event = models.ForeignKey(
to="event.Event", on_delete=models.PROTECT, related_name="submissions"
)
titleCharFieldsource
title = models.CharField(max_length=1000, verbose_name=_("Proposal title"))
submission_typeForeignKeysource
submission_type = models.ForeignKey( # Reasonable default must be set in form/view
to="submission.SubmissionType",
related_name="submissions",
on_delete=models.PROTECT,
verbose_name=_("Session type"),
)
trackForeignKeynullsource
track = models.ForeignKey(
to="submission.Track",
related_name="submissions",
on_delete=models.PROTECT,
verbose_name=_("Track"),
null=True,
blank=True,
)
stateCharFieldsource
state = models.CharField(
max_length=SubmissionStates.get_max_length(),
choices=SubmissionStates.choices,
default=SubmissionStates.SUBMITTED,
verbose_name=_("Proposal state"),
)
pending_stateCharFieldnullsource
pending_state = models.CharField(
null=True,
blank=True,
max_length=SubmissionStates.get_max_length(),
choices=SubmissionStates.choices,
default=None,
verbose_name=_("Pending proposal state"),
)
abstractMarkdownFieldnullsource
abstract = MarkdownField(null=True, blank=True, verbose_name=_("Abstract"))
descriptionMarkdownFieldnullsource
description = MarkdownField(null=True, blank=True, verbose_name=_("Description"))
notesMarkdownFieldnullsourceThese notes are meant for the organisers and won’t be made public.
notes = MarkdownField(
null=True,
blank=True,
verbose_name=_("Notes"),
help_text=_(
"These notes are meant for the organisers and won’t be made public."
),
)
internal_notesTextFieldnullsourceInternal notes for other organisers/reviewers. Not visible to the speakers or the public.
internal_notes = models.TextField(
null=True,
blank=True,
verbose_name=phrases.base.internal_notes,
help_text=phrases.base.internal_notes_help,
)
durationPositiveIntegerFieldnullsourceDuration in minutes
duration = models.PositiveIntegerField(
null=True,
blank=True,
verbose_name=_("Duration"),
help_text=_("Duration in minutes"),
)
slot_countIntegerFieldsourceHow often this session takes place.
slot_count = models.IntegerField(
default=1,
verbose_name=_("Slot count"),
help_text=_("How often this session takes place."),
validators=[MinValueValidator(1)],
)
attendee_signup_requiredBooleanFieldnullsourceOverride whether attendees must sign up to attend this session.
attendee_signup_required = models.BooleanField(
null=True, # None means that the track and submission_type settings are used
blank=True,
verbose_name=_("Requires signup"),
help_text=_("Override whether attendees must sign up to attend this session."),
)
attendee_signup_capacityPositiveIntegerFieldnullsourceOverride the room capacity for this session.
attendee_signup_capacity = models.PositiveIntegerField(
null=True,
blank=True,
verbose_name=_("Attendee capacity"),
help_text=_("Override the room capacity for this session."),
validators=[MinValueValidator(1)],
)
content_localeCharFieldsource
content_locale = models.CharField(
max_length=32, default=settings.LANGUAGE_CODE, verbose_name=_("Language")
)
is_featuredBooleanFieldsource
is_featured = models.BooleanField(
default=False,
verbose_name=_("Show this session in public list of featured sessions."),
)
do_not_recordBooleanFieldsource
do_not_record = models.BooleanField(
default=False, verbose_name=_("Don’t record this session.")
)
imageImageFieldnullsourceUse this if you want an illustration to go with your proposal.
image = models.ImageField(
null=True,
blank=True,
upload_to=submission_image_path,
verbose_name=_("Session image"),
help_text=phrases.base.image_help,
)
invitation_tokenCharFieldsource
invitation_token = models.CharField(max_length=32, default=generate_invite_code)
access_codeForeignKeynullsource
access_code = models.ForeignKey(
to="submission.SubmitterAccessCode",
related_name="submissions",
on_delete=models.PROTECT,
null=True,
blank=True,
)
review_codeCharFieldnulluniquesource
review_code = models.CharField(
max_length=32, unique=True, null=True, blank=True, default=generate_invite_code
)
anonymisedJSONFieldnullsource
anonymised = models.JSONField(null=True, blank=True)
draft_additional_speakersJSONFieldsource
draft_additional_speakers = models.JSONField(default=list, blank=True)
createdDateTimeFieldnullTimestampedModelsource
on pretalx.common.models.mixins.TimestampedModel source
created = models.DateTimeField(
verbose_name=_("Created"), auto_now_add=True, blank=True, null=True
)
updatedDateTimeFieldnullTimestampedModelsource
on pretalx.common.models.mixins.TimestampedModel source
updated = models.DateTimeField(
verbose_name=_("Updated"), auto_now=True, blank=True, null=True
)
Reverse relations 14
Attributes 3
all_objects<django_scopes.manager.ScopedManager.<locals>.Manager>log_prefix'pretalx.submission'objects<django_scopes.manager.ScopedManager.<locals>.Manager>Inherited attributes 5
code_charset['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'J', 'K', 'L', 'M', 'N', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X', 'Y', 'Z', '3', '7', '8', '9']GenerateCodecode_length6GenerateCodecode_property'code'GenerateCodecode_scope()GenerateCodelog_parentNoneLogMixinProperties 31
Defined here
active_resourcessource
@cached_property
def active_resources(self):
resources = [
resource
for resource in self.resources.all()
if resource.link or (resource.resource and resource.resource.name != "None")
]
resources.sort(key=lambda resource: resource.link or "")
return resources
availabilitiessourceThe intersection of all.
Availability objects of all speakers of this submission.
@property
def availabilities(self):
"""The intersection of all.
:class:`~pretalx.schedule.models.availability.Availability` objects of
all speakers of this submission.
"""
all_availabilities = self.event.valid_availabilities.filter(
person__in=self.speakers.all()
)
return Availability.intersection(all_availabilities)
confirmed_signup_countsource
@cached_property
def confirmed_signup_count(self) -> int:
annotated = getattr(self, "_annotated_confirmed_signup_count", None)
if annotated is not None:
return annotated
return self.attendee_signups.filter(
state=AttendeeSignupStates.CONFIRMED
).count()
current_slotssource
@cached_property
def current_slots(self):
if not self.event.current_schedule:
return None
return self.event.current_schedule.talks.filter(
submission=self, is_visible=True
).select_related("room", "submission", "submission__event")
display_speaker_namessourceHelper method for a consistent speaker name display.
@cached_property
def display_speaker_names(self):
"""Helper method for a consistent speaker name display."""
return ", ".join(s.get_display_name() for s in self.sorted_speakers)
display_title_with_speakerssource
@cached_property
def display_title_with_speakers(self):
title = (
f"{phrases.base.quotation_open}{self.title}{phrases.base.quotation_close}"
)
if not self.sorted_speakers:
return title
return _("{title_in_quotes} by {list_of_speakers}").format(
title_in_quotes=title, list_of_speakers=self.display_speaker_names
)
does_accept_feedbacksource
@cached_property
def does_accept_feedback(self):
slot = self.slot
if slot and slot.start:
return slot.start < now()
return False
editablesourceChecks if the speaker is currently allowed to edit the submission.
@cached_property
def editable(self) -> bool:
"""
Checks if the speaker is currently allowed to edit the submission.
"""
try:
event = self.event
except ObjectDoesNotExist:
# Unsaved submissions can always be edited
return True
deadline = self.submission_type.deadline or event.cfp.deadline
deadline_open = (not deadline) or now() <= deadline
if self.state == SubmissionStates.DRAFT:
# We have to check if we comply with the standard submission requirements if
# we are in a draft state, as drafts should only be editable when they could
# also be submitted.
# For existing drafts with access codes, we ignore the redemption count
# since the code was already redeemed when creating the draft.
access_code = (
self.access_code
if (self.access_code and self.access_code.time_valid)
else None
)
if (self.track and self.track.requires_access_code) and not access_code:
return False
if self.submission_type.requires_access_code and not access_code:
return False
# We are not missing an access code, so we can just check if we hit the
# deadline or can ignore it safely
return bool(deadline_open or access_code)
if not event.get_feature_flag("speakers_can_edit_submissions"):
return False
if self.state == SubmissionStates.SUBMITTED:
return deadline_open or (
event.active_review_phase
and event.active_review_phase.speakers_can_change_submissions
)
return self.state in SubmissionStates.accepted_states
effective_signup_capacitysource
@cached_property
def effective_signup_capacity(self) -> int | None:
if hasattr(self, "_annotated_signup_capacity"):
return self._annotated_signup_capacity
if self.attendee_signup_capacity is not None:
return self.attendee_signup_capacity
slot = self.slot
if slot and slot.room:
return slot.room.capacity
return None
export_durationsource
@cached_property
def export_duration(self):
return serialize_duration(minutes=self.get_duration())
image_urlsource
@property
def image_url(self):
return self.image.url if self.image else ""
integer_uuidsource
@cached_property
def integer_uuid(self):
# For import into Engelsystem, we need to somehow convert our submission code into an unique integer. Luckily,
# codes can contain 34 different characters (including compatibility with frab imported data) and normally have
# 6 charactes. Since log2(34 **6) == 30.52, that just fits in to a positive 32-bit signed integer (that
# Engelsystem expects), if we do it correctly.
charset = [
*self.code_charset,
"1",
"2",
"4",
"5",
"6",
"0",
] # compatibility with imported frab data
base = len(charset)
table = {char: cp for cp, char in enumerate(charset)}
intval = 0
for char in self.code:
intval *= base
intval += table[char]
return intval
is_anonymisedsource
@property
def is_anonymised(self) -> bool:
if self.anonymised:
return bool(self.anonymised.get("_anonymised", False))
return False
is_oversource
@cached_property
def is_over(self):
ends = [slot.real_end for slot in self.current_slots or () if slot.start]
if not ends:
return False
return max(ends) < now()
log_parentsource
@property
def log_parent(self):
return self.event
mean_scoresource
@cached_property
def mean_score(self) -> float | None:
scores = [
review.score for review in self.reviews.all() if review.score is not None
]
return round(statistics.fmean(scores), 1) if scores else None
median_scoresource
@cached_property
def median_score(self) -> float | None:
scores = [
review.score for review in self.reviews.all() if review.score is not None
]
return statistics.median(scores) if scores else None
private_resourcessource
@cached_property
def private_resources(self):
return [
resource for resource in self.active_resources if not resource.is_public
]
public_answerssource
@cached_property
def public_answers(self):
from pretalx.submission.domain.queries.question import ( # noqa: PLC0415 -- thin method
public_answers_for_submission,
)
return public_answers_for_submission(self)
public_resourcessource
@cached_property
def public_resources(self):
return [resource for resource in self.active_resources if resource.is_public]
public_review_link_activesource
@cached_property
def public_review_link_active(self) -> bool:
return (
bool(self.review_code)
and self.state in SubmissionStates.public_review_states
and self.event.get_feature_flag("submission_public_review")
)
public_slotssourceAll publicly visible TalkSlot objects of this submission in the current.
@cached_property
def public_slots(self):
"""All publicly visible :class:`~pretalx.schedule.models.slot.TalkSlot`
objects of this submission in the current.
:class:`~pretalx.schedule.models.schedule.Schedule`.
"""
if not agenda_rules.is_agenda_visible(None, self.event):
return []
return self.current_slots
requires_signupsource
@cached_property
def requires_signup(self) -> bool:
annotated = getattr(self, "_annotated_requires_signup", None)
if annotated is not None:
return annotated
if self.attendee_signup_required is not None:
return self.attendee_signup_required
track_requires = bool(self.track_id and self.track.attendee_signup_required)
if not self.submission_type_id:
# Unsaved instance may not have a submission type yet
return track_requires
return track_requires or self.submission_type.attendee_signup_required
reviewer_answerssource
@cached_property
def reviewer_answers(self):
return self.answers.filter(question__is_visible_to_reviewers=True).order_by(
"question__position"
)
score_categoriessource
@cached_property
def score_categories(self):
track = self.track
track_filter = models.Q(limit_tracks__isnull=True)
if track:
track_filter |= models.Q(limit_tracks__in=[track])
return self.event.score_categories.filter(track_filter, active=True).order_by(
"id"
)
signup_capacity_percentsource
@cached_property
def signup_capacity_percent(self) -> int | None:
capacity = self.effective_signup_capacity
if not capacity:
return None
return min(100, round(self.confirmed_signup_count * 100 / capacity))
signup_places_leftsource
@cached_property
def signup_places_left(self) -> int | None:
capacity = self.effective_signup_capacity
if capacity is None:
return None
return max(capacity - self.confirmed_signup_count, 0)
signup_statussource
@cached_property
def signup_status(self) -> str | None:
if hasattr(self, "_annotated_signup_status"):
return self._annotated_signup_status
if not self.event.get_feature_flag("attendee_signup"):
return None
if not self.requires_signup:
return None
capacity = self.effective_signup_capacity
if capacity is not None and self.confirmed_signup_count >= capacity:
return SignupStatus.FULL
return SignupStatus.OPEN
slotsourceThe first scheduled TalkSlot of this submission in the current.
Note that this slot is not guaranteed to be visible.
@cached_property
def slot(self):
"""The first scheduled :class:`~pretalx.schedule.models.slot.TalkSlot`
of this submission in the current.
:class:`~pretalx.schedule.models.schedule.Schedule`.
Note that this slot is not guaranteed to be visible.
"""
return (
self.event.current_schedule.talks.filter(submission=self)
.select_related("room", "submission", "submission__event")
.first()
if self.event.current_schedule
else None
)
sorted_speakerssource
@cached_property
def sorted_speakers(self):
if "speakers" in getattr(self, "_prefetched_objects_cache", {}):
return self.speakers.all()
return self.speakers.select_related("user", "event").order_by(
"speaker_roles__position"
)
user_statesource
@property
def user_state(self):
deadline = self.submission_type.deadline or self.event.cfp.deadline
cfp_open = (not deadline) or now() <= deadline
if self.state == SubmissionStates.SUBMITTED and not cfp_open:
return "review"
return self.state
Methods 45
Defined here
accept(person=None, orga: bool = True)source
def accept(self, person=None, orga: bool = True):
from pretalx.submission.domain.submission import ( # noqa: PLC0415 -- thin method
set_submission_state,
)
set_submission_state(self, SubmissionStates.ACCEPTED, person=person, orga=orga)
add_favourite(user)source
def add_favourite(self, user):
SubmissionFavourite.objects.get_or_create(user=user, submission=self)
cancel(person=None, orga: bool = True)source
def cancel(self, person=None, orga: bool = True):
from pretalx.submission.domain.submission import ( # noqa: PLC0415 -- thin method
set_submission_state,
)
set_submission_state(self, SubmissionStates.CANCELED, person=person, orga=orga)
clean()sourceHook for doing any extra model-wide validation after clean() has been called on every field by self.clean_fields.
on pretalx.submission.models.submission.Submission source
Any ValidationError raised by this method will not be associated with a particular field; it will have a special-case association with the field defined by NON_FIELD_ERRORS.
def clean(self):
super().clean()
# The model field default is settings.LANGUAGE_CODE, which can disagree
# with the event's locale (e.g. a German event on a server defaulting to
# English). Forms/serializers may not surface the field at all when the
# event has a single content locale, so the fallback lives on the model.
if self.event_id and self.content_locale not in self.event.content_locales:
self.content_locale = self.event.locale
validate_signup_required(self, self.attendee_signup_required)
Hook for doing any extra model-wide validation after clean() has been called on every field by self.clean_fields. Any ValidationError raised by this method will not be associated with a particular field; it will have a special-case association with the field defined by NON_FIELD_ERRORS.
def clean(self):
"""
Hook for doing any extra model-wide validation after clean() has been
called on every field by self.clean_fields. Any ValidationError raised
by this method will not be associated with a particular field; it will
have a special-case association with the field defined by
NON_FIELD_ERRORS.
"""
pass
confirm(person=None, orga: bool = False)source
def confirm(self, person=None, orga: bool = False):
from pretalx.submission.domain.submission import ( # noqa: PLC0415 -- thin method
set_submission_state,
)
set_submission_state(self, SubmissionStates.CONFIRMED, person=person, orga=orga)
get_anonymised(attribute)source
def get_anonymised(self, attribute):
if self.is_anonymised and attribute in self.anonymised:
return self.anonymised[attribute]
return getattr(self, attribute, None)
get_content_locale_display()source
def get_content_locale_display(self):
locale_names = dict(self.event.named_content_locales)
if self.content_locale not in locale_names:
locale_names = dict(self.event.available_content_locales)
return str(locale_names.get(self.content_locale, self.content_locale))
get_duration() -> intsource
def get_duration(self) -> int:
if self.duration is None: # We permit zero-length duration
return self.submission_type.default_duration
return self.duration
get_email_locale(fallback=None)source
def get_email_locale(self, fallback=None):
if self.content_locale in self.event.locales:
return self.content_locale
if fallback and fallback in self.event.locales:
return fallback
return self.event.locale
get_instance_data()sourceGet a dictionary of field values for this instance.
on pretalx.submission.models.submission.Submission source
Used for change tracking in log_action. Excludes auto-updated fields like timestamps and sensitive data. Does not handle many-to-many fields.
def get_instance_data(self):
data = super().get_instance_data()
if not self._state.adding:
lines = [line for r in self.resources.all() if (line := r.as_markdown)]
if lines:
data["resources"] = "\n".join(f"- {line}" for line in lines)
tags = list(self.tags.values_list("tag", flat=True)) or []
data["tags"] = "\n".join(f"- {tag}" for tag in tags)
return data
on pretalx.common.models.mixins.LogMixin source
Get a dictionary of field values for this instance.
Used for change tracking in log_action. Excludes auto-updated fields like timestamps and sensitive data. Does not handle many-to-many fields.
def get_instance_data(self):
"""Get a dictionary of field values for this instance.
Used for change tracking in log_action. Excludes auto-updated
fields like timestamps and sensitive data.
Does not handle many-to-many fields.
"""
excluded_fields = {
"created",
"updated",
"is_active",
"last_login",
"user",
"event",
"code",
}
data = {}
for field in self._meta.fields:
if (
field.name in excluded_fields
or field.name in SENSITIVE_KEYS
or "thumbnail" in field.name
or getattr(field, "auto_now", False)
or getattr(field, "auto_now_add", False)
):
continue
if isinstance(field, models.ForeignKey):
data[field.name] = getattr(self, field.attname, None)
continue
value = getattr(self, field.name, None)
if isinstance(field, models.FileField):
data[field.name] = value.name if value else None
elif isinstance(field, models.UUIDField):
data[field.name] = str(value) if value else None
elif isinstance(value, LazyI18nString):
if isinstance(getattr(value, "data", None), dict):
data[field.name] = {k: v for k, v in value.data.items() if v}
else:
data[field.name] = str(value)
else:
data[field.name] = json_roundtrip(value)
return data
log_action(action, data=None, **kwargs)source
on pretalx.submission.models.submission.Submission source
def log_action(self, action, data=None, **kwargs):
if self.state != SubmissionStates.DRAFT:
return super().log_action(action=action, data=data, **kwargs)
on pretalx.common.models.mixins.LogMixin source
def log_action(
self,
action,
data=None,
person=None,
orga=False,
content_object=None,
old_data=None,
new_data=None,
):
if self._state.adding or not isinstance(self.pk, int):
return
if action.startswith("."):
if self.log_prefix:
action = f"{self.log_prefix}{action}"
else:
return
if old_data is not None or new_data is not None:
from pretalx.common.log import ( # noqa: PLC0415 -- thin method
compute_log_changes,
)
changes = compute_log_changes(old_data, new_data)
if not changes and not data:
return
if changes:
if data is None:
data = {}
data["changes"] = changes
if data:
if not isinstance(data, dict):
raise TypeError(
f"Logged data should always be a dictionary, not {type(data)}."
)
for key in data:
if any(sensitive_key in key for sensitive_key in SENSITIVE_KEYS):
data[key] = "********" if data[key] else data[key]
data = json_roundtrip(data)
return ActivityLog.objects.create(
person=person,
content_object=content_object or self,
action_type=action,
data=data,
is_orga_action=orga,
**self._log_event_kwargs(),
)
reject(person=None, orga: bool = True)source
def reject(self, person=None, orga: bool = True):
from pretalx.submission.domain.submission import ( # noqa: PLC0415 -- thin method
set_submission_state,
)
set_submission_state(self, SubmissionStates.REJECTED, person=person, orga=orga)
remove_favourite(user)source
def remove_favourite(self, user):
SubmissionFavourite.objects.filter(user=user, submission=self).delete()
withdraw(person=None, orga: bool = False)source
def withdraw(self, person=None, orga: bool = False):
from pretalx.submission.domain.submission import ( # noqa: PLC0415 -- thin method
set_submission_state,
)
set_submission_state(self, SubmissionStates.WITHDRAWN, person=person, orga=orga)
__str__()sourceReturn str(self).
on pretalx.submission.models.submission.Submission source
def __str__(self):
if not self._state.adding:
return f"Submission(event={self.event.slug}, code={self.code}, title={self.title}, state={self.state})"
return f"Submission(code={self.code}, title={self.title}, state={self.state})"
Return str(self).
def __str__(self):
return "%s object (%s)" % (self.__class__.__name__, self.pk)
From GenerateCode pretalx.common.models.mixins
assign_code(length=None)source
def assign_code(self, length=None):
length = length or self.code_length
while True:
code = self.generate_code(length=length)
filter_kwargs = {f"{self.code_property}__iexact": code}
for field in self.code_scope:
filter_kwargs[field] = getattr(self, field)
with scopes_disabled():
if not self.__class__.objects.filter(**filter_kwargs).exists():
setattr(self, self.code_property, code)
return
save(*args, **kwargs)source
on pretalx.common.models.mixins.GenerateCode source
def save(self, *args, **kwargs):
if getattr(self, self.code_property, None):
return super().save(*args, **kwargs)
# Auto-generate code with retry loop to handle unlikely race conditions
if "update_fields" in kwargs:
kwargs["update_fields"] = {self.code_property}.union(
kwargs["update_fields"]
)
for attempt in range(
3
): # pragma: no branch -- loop always exits via return or raise
self.assign_code()
try:
with transaction.atomic():
return super().save(*args, **kwargs)
except IntegrityError:
if attempt == 2:
raise
setattr(self, self.code_property, None)
on pretalx.common.models.mixins.FileCleanupMixin source
def save(self, *args, **kwargs):
update_fields = kwargs.get("update_fields")
if self._state.adding or (
update_fields and not set(self._file_fields) & set(update_fields)
):
return super().save(*args, **kwargs)
try:
pre_save_instance = self.__class__.objects.get(pk=self.pk)
except ObjectDoesNotExist:
return super().save(*args, **kwargs)
old_files = {}
for field in self._file_fields:
if old_value := getattr(pre_save_instance, field):
new_value = getattr(self, field)
if new_value and old_value.path != new_value.path:
old_files[field] = old_value.path
result = super().save(*args, **kwargs)
for field, path in old_files.items():
self._schedule_file_cleanup(field=field, path=path)
return result
Save the current instance. Override this in a subclass if you want to control the saving process.
The 'force_insert' and 'force_update' parameters can be used to insist that the "save" must be an SQL insert or update (or equivalent for non-SQL backends), respectively. Normally, they should not be set.
def save(
self,
*,
force_insert=False,
force_update=False,
using=None,
update_fields=None,
):
"""
Save the current instance. Override this in a subclass if you want to
control the saving process.
The 'force_insert' and 'force_update' parameters can be used to insist
that the "save" must be an SQL insert or update (or equivalent for
non-SQL backends), respectively. Normally, they should not be set.
"""
self._prepare_related_fields_for_save(operation_name="save")
using = using or router.db_for_write(self.__class__, instance=self)
if force_insert and (force_update or update_fields):
raise ValueError("Cannot force both insert and updating in model saving.")
deferred_non_generated_fields = {
f.attname
for f in self._meta.concrete_fields
if f.attname not in self.__dict__ and f.generated is False
}
if update_fields is not None:
# If update_fields is empty, skip the save. We do also check for
# no-op saves later on for inheritance cases. This bailout is
# still needed for skipping signal sending.
if not update_fields:
return
update_fields = frozenset(update_fields)
field_names = self._meta._non_pk_concrete_field_names
not_updatable_fields = update_fields.difference(field_names)
if not_updatable_fields:
raise ValueError(
"The following fields do not exist in this model, are m2m "
"fields, primary keys, or are non-concrete fields: %s"
% ", ".join(not_updatable_fields)
)
# If saving to the same database, and this model is deferred, then
# automatically do an "update_fields" save on the loaded fields.
elif (
not force_insert
and deferred_non_generated_fields
and using == self._state.db
and self._is_pk_set()
):
field_names = set()
pk_fields = self._meta.pk_fields
for field in self._meta.concrete_fields:
if field not in pk_fields and not hasattr(field, "through"):
field_names.add(field.attname)
loaded_fields = field_names.difference(deferred_non_generated_fields)
if loaded_fields:
update_fields = frozenset(loaded_fields)
self.save_base(
using=using,
force_insert=force_insert,
force_update=force_update,
update_fields=update_fields,
)
From LogMixin pretalx.common.models.mixins
delete(*args, log_kwargs=None, skip_log=False, **kwargs)source
on pretalx.common.models.mixins.LogMixin source
def delete(self, *args, log_kwargs=None, skip_log=False, **kwargs):
parent = self.log_parent
result = super().delete(*args, **kwargs)
if (
not skip_log
and parent
and getattr(parent, "log_action", None)
and self.log_prefix
):
log_kwargs = log_kwargs or {}
parent.log_action(f"{self.log_prefix}.delete", **log_kwargs)
return result
on pretalx.common.models.mixins.FileCleanupMixin source
def delete(self, *args, **kwargs):
self.delete_files()
return super().delete(*args, **kwargs)
def delete(self, using=None, keep_parents=False):
if not self._is_pk_set():
raise ValueError(
"%s object can't be deleted because its %s attribute is set "
"to None." % (self._meta.object_name, self._meta.pk.attname)
)
using = using or router.db_for_write(self.__class__, instance=self)
collector = Collector(using=using, origin=self)
collector.collect([self], keep_parents=keep_parents)
return collector.delete()
logged_actions()source
def logged_actions(self):
return (
ActivityLog.objects.filter(
content_type=ContentType.objects.get_for_model(type(self)),
object_id=self.pk,
)
.select_related("event", "person")
.prefetch_related("content_object")
)
From FileCleanupMixin pretalx.common.models.mixins
delete_files()sourceSchedule cleanup of every uploaded file attached to this object.
Public hook: called by delete() and by anonymisation paths (e.g. person.domain.user.deactivate_user) that want to drop files without removing the row. Overridable by subclasses that need to recurse into related objects.
def delete_files(self):
"""Schedule cleanup of every uploaded file attached to this object.
Public hook: called by ``delete()`` and by anonymisation paths
(e.g. ``person.domain.user.deactivate_user``) that want to drop
files without removing the row. Overridable by subclasses that
need to recurse into related objects."""
for field in self._file_fields:
value = getattr(self, field, None)
if not value:
continue
with suppress(Exception):
self._schedule_file_cleanup(field=field, path=value.path)
process_image(field, generate_thumbnail=False)source
def process_image(self, field, generate_thumbnail=False):
task_process_image.apply_async(
kwargs={
"field": field,
"model": self._meta.model_name.capitalize(),
"pk": self.pk,
"generate_thumbnail": generate_thumbnail,
},
countdown=10,
)