TalkSlot
pretalx.schedule.models.slot.TalkSlot source
The TalkSlot object is the scheduled version of a Submission.
TalkSlots always belong to one submission and one Schedule.
TalkSlots are publicly visible if their submission was in the confirmed state at schedule release time. Additionally, TalkSlots of type "break" are always publicly visible, and of type "blocker" are never publicly visible.
Fields 11
idAutoFieldpksubmissionForeignKeynullsource
submission = models.ForeignKey(
to="submission.Submission",
on_delete=models.PROTECT,
related_name="slots",
null=True,
blank=True, # If the submission is empty, this is a break or similar event
)
roomForeignKeynullsourceThe room this session is scheduled in, if any
room = models.ForeignKey(
to="schedule.Room",
on_delete=models.PROTECT,
related_name="talks",
verbose_name=_("Room"),
help_text=_("The room this session is scheduled in, if any"),
null=True,
blank=True,
)
scheduleForeignKeysource
schedule = models.ForeignKey(
to="schedule.Schedule", on_delete=models.PROTECT, related_name="talks"
)
is_visibleBooleanFieldsource
is_visible = models.BooleanField(default=False)
slot_typeCharFieldnullsourceFor non-submission slots: 'break' for public breaks, 'blocker' for hidden blockers
slot_type = models.CharField(
max_length=10,
choices=SlotType.choices,
null=True,
blank=True,
verbose_name=_("Slot type"),
help_text=_(
"For non-submission slots: 'break' for public breaks, 'blocker' for hidden blockers"
),
)
startDateTimeFieldnullsourceWhen the session starts, if it is currently scheduled
start = DateTimeField(
null=True,
verbose_name=_("Start"),
help_text=_("When the session starts, if it is currently scheduled"),
)
endDateTimeFieldnullsourceWhen the session ends, if it is currently scheduled
end = DateTimeField(
null=True,
verbose_name=_("End"),
help_text=_("When the session ends, if it is currently scheduled"),
)
descriptionI18nCharFieldnullsource
description = I18nCharField(null=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
)
Attributes 1
objects<django_scopes.manager.ScopedManager.<locals>.Manager>Inherited attributes 2
log_parentNoneLogMixinlog_prefixNoneLogMixinProperties 12
Defined here
as_availabilitysource'Casts' a slot as Availability, useful for availability arithmetic.
@cached_property
def as_availability(self):
"""'Casts' a slot as :class:`~pretalx.schedule.models.availability.Availability`,
useful for availability arithmetic.
"""
return Availability(start=self.start, end=self.real_end)
durationsourceReturns the actual duration in minutes if the talk is scheduled, and the planned duration in minutes otherwise.
@property
def duration(self) -> int:
"""Returns the actual duration in minutes if the talk is scheduled, and
the planned duration in minutes otherwise."""
if self.start and self.end:
return int((self.end - self.start).total_seconds() / 60)
if not self.submission:
return None
return self.submission.get_duration()
eventsource
@cached_property
def event(self):
return self.submission.event if self.submission else self.schedule.event
export_durationsource
@cached_property
def export_duration(self):
return serialize_duration(minutes=self.duration)
frab_slugsource
@cached_property
def frab_slug(self):
title = re.sub(WHITESPACE_REGEX, "-", self.submission.title)
title = title.lower()
title = unicodedata.normalize("NFD", title).encode("ASCII", "ignore").decode()
title = re.sub(FRAB_SLUG_REGEX, "", title)
title = title.strip("-")
if title:
return f"{self.event.slug}-{self.submission.pk}{self.id_suffix}-{title}"
return f"{self.event.slug}-{self.submission.pk}{self.id_suffix}"
id_suffixsource
@cached_property
def id_suffix(self):
if not self.event.get_feature_flag("present_multiple_times"):
return ""
all_slots = list(
TalkSlot.objects.filter(
submission_id=self.submission_id, schedule_id=self.schedule_id
).order_by("start")
)
if len(all_slots) == 1:
return ""
return "-" + str(all_slots.index(self))
local_endsource
@cached_property
def local_end(self):
if self.real_end:
return self.real_end.astimezone(self.event.tz)
local_startsource
@cached_property
def local_start(self):
if self.start:
return self.start.astimezone(self.event.tz)
pentabarf_export_durationsource
@cached_property
def pentabarf_export_duration(self):
duration = dt.timedelta(minutes=self.duration)
days = duration.days
hours = int(duration.total_seconds() // 3600 - days * 24)
minutes = duration.seconds // 60 % 60
return f"{hours:02}{minutes:02}00"
real_endsourceGuaranteed to provide a useful end datetime if start is set, even if end is empty.
@cached_property
def real_end(self):
"""Guaranteed to provide a useful end datetime if ``start`` is set,
even if ``end`` is empty."""
return self.end or (
self.start + dt.timedelta(minutes=self.duration) if self.start else None
)
signup_statussource
@cached_property
def signup_status(self) -> str | None:
if hasattr(self, "_annotated_signup_status"):
return self._annotated_signup_status
if not self.submission_id:
return None
return self.submission.signup_status
uuidsourceA UUID5, calculated from the submission code and the instance identifier.
@cached_property
def uuid(self):
"""A UUID5, calculated from the submission code and the instance identifier."""
global INSTANCE_IDENTIFIER # noqa: PLW0603 -- module-level cache for instance identifier
if not INSTANCE_IDENTIFIER:
INSTANCE_IDENTIFIER = GlobalSettings().get_instance_identifier()
return uuid.uuid5(INSTANCE_IDENTIFIER, self.submission.code + self.id_suffix)
Methods 33
Defined here
clean()sourceHook for doing any extra model-wide validation after clean() has been called on every field by self.clean_fields.
on pretalx.schedule.models.slot.TalkSlot 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()
event = self.event
errors = {}
for field in ("start", "end"):
try:
validate_slot_within_event(getattr(self, field), event=event)
except ValidationError as exc:
errors[field] = exc.messages
try:
validate_slot_time_range(start=self.start, end=self.end)
except ValidationError as exc:
errors.setdefault("end", []).extend(exc.messages)
if self.room_id and self.room.hidden:
errors["room"] = [ROOM_HIDDEN_ERROR]
if errors:
raise ValidationError(errors)
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
is_same_slot(other_slot) -> boolsourceChecks if both slots have the same room and start time.
def is_same_slot(self, other_slot) -> bool:
"""Checks if both slots have the same room and start time."""
return self.room == other_slot.room and self.start == other_slot.start
__str__()sourceReturn str(self).
on pretalx.schedule.models.slot.TalkSlot source
def __str__(self):
return f"TalkSlot(event={self.schedule.event.slug}, submission={getattr(self.submission, 'title', None)}, schedule={self.schedule.version})"
Return str(self).
def __str__(self):
return "%s object (%s)" % (self.__class__.__name__, self.pk)
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()
get_instance_data()sourceGet 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, person=None, orga=False, content_object=None, old_data=None, new_data=None)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(),
)
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,
)
save(*args, **kwargs)source
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,
)