Module fl_server_core.models¶
View Source
# SPDX-FileCopyrightText: 2024 Benedikt Franke <benedikt.franke@dlr.de>
# SPDX-FileCopyrightText: 2024 Florian Heinrich <florian.heinrich@dlr.de>
#
# SPDX-License-Identifier: Apache-2.0
from .metric import Metric
from .model import GlobalModel, LocalModel, MeanModel, Model, SWAGModel
from .training import Training
from .user import User
__all__ = [
"GlobalModel",
"LocalModel",
"Metric",
"MeanModel",
"Model",
"SWAGModel",
"Training",
"User",
]
Sub-modules¶
- fl_server_core.models.metric
- fl_server_core.models.model
- fl_server_core.models.training
- fl_server_core.models.user
Classes¶
GlobalModel¶
Model class for global models.
View Source
class GlobalModel(Model):
"""
Model class for global models.
"""
name: CharField = CharField(max_length=256)
"""Name of the model."""
description: TextField = TextField()
"""Description of the model."""
# alternative to be postgres independent: create a new model for each nullable integer field
# and map the corresponding list of integers to the model (but pay attention to the order)
input_shape: ArrayField = ArrayField(IntegerField(null=True), null=True)
"""Input shape of the model."""
preprocessing: BinaryField = BinaryField(null=True)
"""Preprocessing of the model."""
def get_preprocessing_torch_model(self) -> torch.nn.Module:
"""
Converts the preprocessing to a PyTorch model.
Returns:
torch.nn.Module: The PyTorch model.
"""
return to_torch_module(self.preprocessing)
def set_preprocessing_torch_model(self, value: torch.nn.Module):
"""
Sets the preprocessing from a PyTorch model.
Args:
value (torch.nn.Module): The PyTorch model.
"""
self.preprocessing = from_torch_module(value)
Ancestors (in MRO)¶
- fl_server_core.models.Model
- polymorphic.models.PolymorphicModel
- django.db.models.base.Model
Descendants¶
- fl_server_core.models.SWAGModel
- fl_server_core.models.MeanModel
Class variables¶
Static methods¶
check¶
View Source
@classmethod
def check(cls, **kwargs):
errors = [
*cls._check_swappable(),
*cls._check_model(),
*cls._check_managers(**kwargs),
]
if not cls._meta.swapped:
databases = kwargs.get("databases") or []
errors += [
*cls._check_fields(**kwargs),
*cls._check_m2m_through_same_relationship(),
*cls._check_long_column_names(databases),
]
clash_errors = (
*cls._check_id_field(),
*cls._check_field_name_clashes(),
*cls._check_model_name_db_lookup_clashes(),
*cls._check_property_name_related_field_accessor_clashes(),
*cls._check_single_primary_key(),
)
errors.extend(clash_errors)
# If there are field name clashes, hide consequent column name
# clashes.
if not clash_errors:
errors.extend(cls._check_column_name_clashes())
errors += [
*cls._check_index_together(),
*cls._check_unique_together(),
*cls._check_indexes(databases),
*cls._check_ordering(),
*cls._check_constraints(databases),
*cls._check_default_pk(),
]
return errors
from_db¶
View Source
@classmethod
def from_db(cls, db, field_names, values):
if len(values) != len(cls._meta.concrete_fields):
values_iter = iter(values)
values = [
next(values_iter) if f.attname in field_names else DEFERRED
for f in cls._meta.concrete_fields
]
new = cls(*values)
new._state.adding = False
new._state.db = db
return new
translate_polymorphic_Q_object¶
View Source
Instance variables¶
Methods¶
clean¶
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.
View Source
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
clean_fields¶
Clean all fields and raise a ValidationError containing a dict
of all validation errors if any occur.
View Source
def clean_fields(self, exclude=None):
"""
Clean all fields and raise a ValidationError containing a dict
of all validation errors if any occur.
"""
if exclude is None:
exclude = []
errors = {}
for f in self._meta.fields:
if f.name in exclude:
continue
# Skip validation for empty fields with blank=True. The developer
# is responsible for making sure they have a valid value.
raw_value = getattr(self, f.attname)
if f.blank and raw_value in f.empty_values:
continue
try:
setattr(self, f.attname, f.clean(raw_value, self))
except ValidationError as e:
errors[f.name] = e.error_list
if errors:
raise ValidationError(errors)
date_error_message¶
View Source
def date_error_message(self, lookup_type, field_name, unique_for):
opts = self._meta
field = opts.get_field(field_name)
return ValidationError(
message=field.error_messages["unique_for_date"],
code="unique_for_date",
params={
"model": self,
"model_name": capfirst(opts.verbose_name),
"lookup_type": lookup_type,
"field": field_name,
"field_label": capfirst(field.verbose_name),
"date_field": unique_for,
"date_field_label": capfirst(opts.get_field(unique_for).verbose_name),
},
)
delete¶
View Source
def delete(self, using=None, keep_parents=False):
if self.pk is None:
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)
collector.collect([self], keep_parents=keep_parents)
return collector.delete()
description¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
full_clean¶
Call clean_fields(), clean(), and validate_unique() on the model.
Raise a ValidationError for any errors that occur.
View Source
def full_clean(self, exclude=None, validate_unique=True):
"""
Call clean_fields(), clean(), and validate_unique() on the model.
Raise a ValidationError for any errors that occur.
"""
errors = {}
if exclude is None:
exclude = []
else:
exclude = list(exclude)
try:
self.clean_fields(exclude=exclude)
except ValidationError as e:
errors = e.update_error_dict(errors)
# Form.clean() is run even if other validation fails, so do the
# same with Model.clean() for consistency.
try:
self.clean()
except ValidationError as e:
errors = e.update_error_dict(errors)
# Run unique checks, but only for fields that passed validation.
if validate_unique:
for name in errors:
if name != NON_FIELD_ERRORS and name not in exclude:
exclude.append(name)
try:
self.validate_unique(exclude=exclude)
except ValidationError as e:
errors = e.update_error_dict(errors)
if errors:
raise ValidationError(errors)
get_deferred_fields¶
Return a set containing names of deferred fields for this instance.
View Source
get_preprocessing_torch_model¶
Converts the preprocessing to a PyTorch model.
Returns:
| Type | Description |
|---|---|
| torch.nn.Module | The PyTorch model. |
View Source
get_real_concrete_instance_class¶
View Source
get_real_concrete_instance_class_id¶
View Source
get_real_instance¶
Upcast an object to it's actual type.
If a non-polymorphic manager (like base_objects) has been used to retrieve objects, then the complete object with it's real class/type and all fields may be retrieved with this method.
.. note::
Each method call executes one db query (if necessary).
Use the :meth:~polymorphic.managers.PolymorphicQuerySet.get_real_instances
to upcast a complete list in a single efficient query.
View Source
def get_real_instance(self):
"""
Upcast an object to it's actual type.
If a non-polymorphic manager (like base_objects) has been used to
retrieve objects, then the complete object with it's real class/type
and all fields may be retrieved with this method.
.. note::
Each method call executes one db query (if necessary).
Use the :meth:`~polymorphic.managers.PolymorphicQuerySet.get_real_instances`
to upcast a complete list in a single efficient query.
"""
real_model = self.get_real_instance_class()
if real_model == self.__class__:
return self
return real_model.objects.db_manager(self._state.db).get(pk=self.pk)
get_real_instance_class¶
Return the actual model type of the object.
If a non-polymorphic manager (like base_objects) has been used to retrieve objects, then the real class/type of these objects may be determined using this method.
View Source
def get_real_instance_class(self):
"""
Return the actual model type of the object.
If a non-polymorphic manager (like base_objects) has been used to
retrieve objects, then the real class/type of these objects may be
determined using this method.
"""
if self.polymorphic_ctype_id is None:
raise PolymorphicTypeUndefined(
(
"The model {}#{} does not have a `polymorphic_ctype_id` value defined.\n"
"If you created models outside polymorphic, e.g. through an import or migration, "
"make sure the `polymorphic_ctype_id` field points to the ContentType ID of the model subclass."
).format(self.__class__.__name__, self.pk)
)
# the following line would be the easiest way to do this, but it produces sql queries
# return self.polymorphic_ctype.model_class()
# so we use the following version, which uses the ContentType manager cache.
# Note that model_class() can return None for stale content types;
# when the content type record still exists but no longer refers to an existing model.
model = (
ContentType.objects.db_manager(self._state.db)
.get_for_id(self.polymorphic_ctype_id)
.model_class()
)
# Protect against bad imports (dumpdata without --natural) or other
# issues missing with the ContentType models.
if (
model is not None
and not issubclass(model, self.__class__)
and (
self.__class__._meta.proxy_for_model is None
or not issubclass(model, self.__class__._meta.proxy_for_model)
)
):
raise PolymorphicTypeInvalid(
"ContentType {} for {} #{} does not point to a subclass!".format(
self.polymorphic_ctype_id, model, self.pk
)
)
return model
get_torch_model¶
Converts the model weights to a PyTorch model.
Returns:
| Type | Description |
|---|---|
| torch.nn.Module | The PyTorch model. |
View Source
get_training¶
Gets the training associated with the model.
Returns:
| Type | Description |
|---|---|
| models.Training | The training associated with the model. |
View Source
id¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
input_shape¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
is_global_model¶
Checks if the model is a global model.
Returns:
| Type | Description |
|---|---|
| bool | True if the model is a global model, False otherwise. |
View Source
is_local_model¶
Checks if the model is a local model.
Returns:
| Type | Description |
|---|---|
| bool | True if the model is a local model, False otherwise. |
View Source
name¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
pre_save_polymorphic¶
Make sure the polymorphic_ctype value is correctly set on this model.
View Source
def pre_save_polymorphic(self, using=DEFAULT_DB_ALIAS):
"""
Make sure the ``polymorphic_ctype`` value is correctly set on this model.
"""
# This function may be called manually in special use-cases. When the object
# is saved for the first time, we store its real class in polymorphic_ctype.
# When the object later is retrieved by PolymorphicQuerySet, it uses this
# field to figure out the real class of this object
# (used by PolymorphicQuerySet._get_real_instances)
if not self.polymorphic_ctype_id:
self.polymorphic_ctype = ContentType.objects.db_manager(using).get_for_model(
self, for_concrete_model=False
)
prepare_database_save¶
View Source
preprocessing¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
refresh_from_db¶
Reload field values from the database.
By default, the reloading happens from the database this instance was loaded from, or by the read router if this instance wasn't loaded from any database. The using parameter will override the default.
Fields can be used to specify which fields to reload. The fields should be an iterable of field attnames. If fields is None, then all non-deferred fields are reloaded.
When accessing deferred fields of an instance, the deferred loading of the field will call this method.
View Source
def refresh_from_db(self, using=None, fields=None):
"""
Reload field values from the database.
By default, the reloading happens from the database this instance was
loaded from, or by the read router if this instance wasn't loaded from
any database. The using parameter will override the default.
Fields can be used to specify which fields to reload. The fields
should be an iterable of field attnames. If fields is None, then
all non-deferred fields are reloaded.
When accessing deferred fields of an instance, the deferred loading
of the field will call this method.
"""
if fields is None:
self._prefetched_objects_cache = {}
else:
prefetched_objects_cache = getattr(self, "_prefetched_objects_cache", ())
for field in fields:
if field in prefetched_objects_cache:
del prefetched_objects_cache[field]
fields.remove(field)
if not fields:
return
if any(LOOKUP_SEP in f for f in fields):
raise ValueError(
'Found "%s" in fields argument. Relations and transforms '
"are not allowed in fields." % LOOKUP_SEP
)
hints = {"instance": self}
db_instance_qs = self.__class__._base_manager.db_manager(
using, hints=hints
).filter(pk=self.pk)
# Use provided fields, if not set then reload all non-deferred fields.
deferred_fields = self.get_deferred_fields()
if fields is not None:
fields = list(fields)
db_instance_qs = db_instance_qs.only(*fields)
elif deferred_fields:
fields = [
f.attname
for f in self._meta.concrete_fields
if f.attname not in deferred_fields
]
db_instance_qs = db_instance_qs.only(*fields)
db_instance = db_instance_qs.get()
non_loaded_fields = db_instance.get_deferred_fields()
for field in self._meta.concrete_fields:
if field.attname in non_loaded_fields:
# This field wasn't refreshed - skip ahead.
continue
setattr(self, field.attname, getattr(db_instance, field.attname))
# Clear cached foreign keys.
if field.is_relation and field.is_cached(self):
field.delete_cached_value(self)
# Clear cached relations.
for field in self._meta.related_objects:
if field.is_cached(self):
field.delete_cached_value(self)
self._state.db = db_instance._state.db
round¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
save¶
Calls :meth:pre_save_polymorphic and saves the model.
View Source
save_base¶
def save_base(
self,
raw=False,
force_insert=False,
force_update=False,
using=None,
update_fields=None
)
Handle the parts of saving which should be done only once per save,
yet need to be done in raw saves, too. This includes some sanity checks and signal sending.
The 'raw' argument is telling save_base not to save any parent models and not to do any changes to the values before save. This is used by fixture loading.
View Source
def save_base(
self,
raw=False,
force_insert=False,
force_update=False,
using=None,
update_fields=None,
):
"""
Handle the parts of saving which should be done only once per save,
yet need to be done in raw saves, too. This includes some sanity
checks and signal sending.
The 'raw' argument is telling save_base not to save any parent
models and not to do any changes to the values before save. This
is used by fixture loading.
"""
using = using or router.db_for_write(self.__class__, instance=self)
assert not (force_insert and (force_update or update_fields))
assert update_fields is None or update_fields
cls = origin = self.__class__
# Skip proxies, but keep the origin as the proxy model.
if cls._meta.proxy:
cls = cls._meta.concrete_model
meta = cls._meta
if not meta.auto_created:
pre_save.send(
sender=origin,
instance=self,
raw=raw,
using=using,
update_fields=update_fields,
)
# A transaction isn't needed if one query is issued.
if meta.parents:
context_manager = transaction.atomic(using=using, savepoint=False)
else:
context_manager = transaction.mark_for_rollback_on_error(using=using)
with context_manager:
parent_inserted = False
if not raw:
parent_inserted = self._save_parents(cls, using, update_fields)
updated = self._save_table(
raw,
cls,
force_insert or parent_inserted,
force_update,
using,
update_fields,
)
# Store the database on which the object was saved
self._state.db = using
# Once saved, this is no longer a to-be-added instance.
self._state.adding = False
# Signal that the save is complete
if not meta.auto_created:
post_save.send(
sender=origin,
instance=self,
created=(not updated),
update_fields=update_fields,
raw=raw,
using=using,
)
serializable_value¶
Return the value of the field name for this instance. If the field is
a foreign key, return the id value instead of the object. If there's no Field object with this name on the model, return the model attribute's value.
Used to serialize a field's value (in the serializer, or form output, for example). Normally, you would just access the attribute directly and not use this method.
View Source
def serializable_value(self, field_name):
"""
Return the value of the field name for this instance. If the field is
a foreign key, return the id value instead of the object. If there's
no Field object with this name on the model, return the model
attribute's value.
Used to serialize a field's value (in the serializer, or form output,
for example). Normally, you would just access the attribute directly
and not use this method.
"""
try:
field = self._meta.get_field(field_name)
except FieldDoesNotExist:
return getattr(self, field_name)
return getattr(self, field.attname)
set_preprocessing_torch_model¶
Sets the preprocessing from a PyTorch model.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| value | torch.nn.Module | The PyTorch model. | None |
View Source
set_torch_model¶
Sets the model weights from a PyTorch model.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| value | torch.nn.Module | The PyTorch model. | None |
View Source
unique_error_message¶
View Source
def unique_error_message(self, model_class, unique_check):
opts = model_class._meta
params = {
"model": self,
"model_class": model_class,
"model_name": capfirst(opts.verbose_name),
"unique_check": unique_check,
}
# A unique field
if len(unique_check) == 1:
field = opts.get_field(unique_check[0])
params["field_label"] = capfirst(field.verbose_name)
return ValidationError(
message=field.error_messages["unique"],
code="unique",
params=params,
)
# unique_together
else:
field_labels = [
capfirst(opts.get_field(f).verbose_name) for f in unique_check
]
params["field_labels"] = get_text_list(field_labels, _("and"))
return ValidationError(
message=_("%(model_name)s with this %(field_labels)s already exists."),
code="unique_together",
params=params,
)
validate_unique¶
Check unique constraints on the model and raise ValidationError if any
failed.
View Source
def validate_unique(self, exclude=None):
"""
Check unique constraints on the model and raise ValidationError if any
failed.
"""
unique_checks, date_checks = self._get_unique_checks(exclude=exclude)
errors = self._perform_unique_checks(unique_checks)
date_errors = self._perform_date_checks(date_checks)
for k, v in date_errors.items():
errors.setdefault(k, []).extend(v)
if errors:
raise ValidationError(errors)
weights¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
LocalModel¶
Model class for local models.
View Source
class LocalModel(Model):
"""
Model class for local models.
"""
base_model: ForeignKey = ForeignKey(GlobalModel, on_delete=CASCADE)
"""Base model of the local model."""
sample_size: IntegerField = IntegerField()
"""Sample size of the local model."""
def get_training(self) -> Optional["models.Training"]:
"""
Gets the training associated with the base model.
Returns:
models.Training: The training associated with the base model.
"""
return models.Training.objects.filter(model=self.base_model).first()
Ancestors (in MRO)¶
- fl_server_core.models.Model
- polymorphic.models.PolymorphicModel
- django.db.models.base.Model
Class variables¶
Static methods¶
check¶
View Source
@classmethod
def check(cls, **kwargs):
errors = [
*cls._check_swappable(),
*cls._check_model(),
*cls._check_managers(**kwargs),
]
if not cls._meta.swapped:
databases = kwargs.get("databases") or []
errors += [
*cls._check_fields(**kwargs),
*cls._check_m2m_through_same_relationship(),
*cls._check_long_column_names(databases),
]
clash_errors = (
*cls._check_id_field(),
*cls._check_field_name_clashes(),
*cls._check_model_name_db_lookup_clashes(),
*cls._check_property_name_related_field_accessor_clashes(),
*cls._check_single_primary_key(),
)
errors.extend(clash_errors)
# If there are field name clashes, hide consequent column name
# clashes.
if not clash_errors:
errors.extend(cls._check_column_name_clashes())
errors += [
*cls._check_index_together(),
*cls._check_unique_together(),
*cls._check_indexes(databases),
*cls._check_ordering(),
*cls._check_constraints(databases),
*cls._check_default_pk(),
]
return errors
from_db¶
View Source
@classmethod
def from_db(cls, db, field_names, values):
if len(values) != len(cls._meta.concrete_fields):
values_iter = iter(values)
values = [
next(values_iter) if f.attname in field_names else DEFERRED
for f in cls._meta.concrete_fields
]
new = cls(*values)
new._state.adding = False
new._state.db = db
return new
translate_polymorphic_Q_object¶
View Source
Instance variables¶
Methods¶
clean¶
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.
View Source
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
clean_fields¶
Clean all fields and raise a ValidationError containing a dict
of all validation errors if any occur.
View Source
def clean_fields(self, exclude=None):
"""
Clean all fields and raise a ValidationError containing a dict
of all validation errors if any occur.
"""
if exclude is None:
exclude = []
errors = {}
for f in self._meta.fields:
if f.name in exclude:
continue
# Skip validation for empty fields with blank=True. The developer
# is responsible for making sure they have a valid value.
raw_value = getattr(self, f.attname)
if f.blank and raw_value in f.empty_values:
continue
try:
setattr(self, f.attname, f.clean(raw_value, self))
except ValidationError as e:
errors[f.name] = e.error_list
if errors:
raise ValidationError(errors)
date_error_message¶
View Source
def date_error_message(self, lookup_type, field_name, unique_for):
opts = self._meta
field = opts.get_field(field_name)
return ValidationError(
message=field.error_messages["unique_for_date"],
code="unique_for_date",
params={
"model": self,
"model_name": capfirst(opts.verbose_name),
"lookup_type": lookup_type,
"field": field_name,
"field_label": capfirst(field.verbose_name),
"date_field": unique_for,
"date_field_label": capfirst(opts.get_field(unique_for).verbose_name),
},
)
delete¶
View Source
def delete(self, using=None, keep_parents=False):
if self.pk is None:
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)
collector.collect([self], keep_parents=keep_parents)
return collector.delete()
full_clean¶
Call clean_fields(), clean(), and validate_unique() on the model.
Raise a ValidationError for any errors that occur.
View Source
def full_clean(self, exclude=None, validate_unique=True):
"""
Call clean_fields(), clean(), and validate_unique() on the model.
Raise a ValidationError for any errors that occur.
"""
errors = {}
if exclude is None:
exclude = []
else:
exclude = list(exclude)
try:
self.clean_fields(exclude=exclude)
except ValidationError as e:
errors = e.update_error_dict(errors)
# Form.clean() is run even if other validation fails, so do the
# same with Model.clean() for consistency.
try:
self.clean()
except ValidationError as e:
errors = e.update_error_dict(errors)
# Run unique checks, but only for fields that passed validation.
if validate_unique:
for name in errors:
if name != NON_FIELD_ERRORS and name not in exclude:
exclude.append(name)
try:
self.validate_unique(exclude=exclude)
except ValidationError as e:
errors = e.update_error_dict(errors)
if errors:
raise ValidationError(errors)
get_deferred_fields¶
Return a set containing names of deferred fields for this instance.
View Source
get_real_concrete_instance_class¶
View Source
get_real_concrete_instance_class_id¶
View Source
get_real_instance¶
Upcast an object to it's actual type.
If a non-polymorphic manager (like base_objects) has been used to retrieve objects, then the complete object with it's real class/type and all fields may be retrieved with this method.
.. note::
Each method call executes one db query (if necessary).
Use the :meth:~polymorphic.managers.PolymorphicQuerySet.get_real_instances
to upcast a complete list in a single efficient query.
View Source
def get_real_instance(self):
"""
Upcast an object to it's actual type.
If a non-polymorphic manager (like base_objects) has been used to
retrieve objects, then the complete object with it's real class/type
and all fields may be retrieved with this method.
.. note::
Each method call executes one db query (if necessary).
Use the :meth:`~polymorphic.managers.PolymorphicQuerySet.get_real_instances`
to upcast a complete list in a single efficient query.
"""
real_model = self.get_real_instance_class()
if real_model == self.__class__:
return self
return real_model.objects.db_manager(self._state.db).get(pk=self.pk)
get_real_instance_class¶
Return the actual model type of the object.
If a non-polymorphic manager (like base_objects) has been used to retrieve objects, then the real class/type of these objects may be determined using this method.
View Source
def get_real_instance_class(self):
"""
Return the actual model type of the object.
If a non-polymorphic manager (like base_objects) has been used to
retrieve objects, then the real class/type of these objects may be
determined using this method.
"""
if self.polymorphic_ctype_id is None:
raise PolymorphicTypeUndefined(
(
"The model {}#{} does not have a `polymorphic_ctype_id` value defined.\n"
"If you created models outside polymorphic, e.g. through an import or migration, "
"make sure the `polymorphic_ctype_id` field points to the ContentType ID of the model subclass."
).format(self.__class__.__name__, self.pk)
)
# the following line would be the easiest way to do this, but it produces sql queries
# return self.polymorphic_ctype.model_class()
# so we use the following version, which uses the ContentType manager cache.
# Note that model_class() can return None for stale content types;
# when the content type record still exists but no longer refers to an existing model.
model = (
ContentType.objects.db_manager(self._state.db)
.get_for_id(self.polymorphic_ctype_id)
.model_class()
)
# Protect against bad imports (dumpdata without --natural) or other
# issues missing with the ContentType models.
if (
model is not None
and not issubclass(model, self.__class__)
and (
self.__class__._meta.proxy_for_model is None
or not issubclass(model, self.__class__._meta.proxy_for_model)
)
):
raise PolymorphicTypeInvalid(
"ContentType {} for {} #{} does not point to a subclass!".format(
self.polymorphic_ctype_id, model, self.pk
)
)
return model
get_torch_model¶
Converts the model weights to a PyTorch model.
Returns:
| Type | Description |
|---|---|
| torch.nn.Module | The PyTorch model. |
View Source
get_training¶
Gets the training associated with the base model.
Returns:
| Type | Description |
|---|---|
| models.Training | The training associated with the base model. |
View Source
id¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
is_global_model¶
Checks if the model is a global model.
Returns:
| Type | Description |
|---|---|
| bool | True if the model is a global model, False otherwise. |
View Source
is_local_model¶
Checks if the model is a local model.
Returns:
| Type | Description |
|---|---|
| bool | True if the model is a local model, False otherwise. |
View Source
pre_save_polymorphic¶
Make sure the polymorphic_ctype value is correctly set on this model.
View Source
def pre_save_polymorphic(self, using=DEFAULT_DB_ALIAS):
"""
Make sure the ``polymorphic_ctype`` value is correctly set on this model.
"""
# This function may be called manually in special use-cases. When the object
# is saved for the first time, we store its real class in polymorphic_ctype.
# When the object later is retrieved by PolymorphicQuerySet, it uses this
# field to figure out the real class of this object
# (used by PolymorphicQuerySet._get_real_instances)
if not self.polymorphic_ctype_id:
self.polymorphic_ctype = ContentType.objects.db_manager(using).get_for_model(
self, for_concrete_model=False
)
prepare_database_save¶
View Source
refresh_from_db¶
Reload field values from the database.
By default, the reloading happens from the database this instance was loaded from, or by the read router if this instance wasn't loaded from any database. The using parameter will override the default.
Fields can be used to specify which fields to reload. The fields should be an iterable of field attnames. If fields is None, then all non-deferred fields are reloaded.
When accessing deferred fields of an instance, the deferred loading of the field will call this method.
View Source
def refresh_from_db(self, using=None, fields=None):
"""
Reload field values from the database.
By default, the reloading happens from the database this instance was
loaded from, or by the read router if this instance wasn't loaded from
any database. The using parameter will override the default.
Fields can be used to specify which fields to reload. The fields
should be an iterable of field attnames. If fields is None, then
all non-deferred fields are reloaded.
When accessing deferred fields of an instance, the deferred loading
of the field will call this method.
"""
if fields is None:
self._prefetched_objects_cache = {}
else:
prefetched_objects_cache = getattr(self, "_prefetched_objects_cache", ())
for field in fields:
if field in prefetched_objects_cache:
del prefetched_objects_cache[field]
fields.remove(field)
if not fields:
return
if any(LOOKUP_SEP in f for f in fields):
raise ValueError(
'Found "%s" in fields argument. Relations and transforms '
"are not allowed in fields." % LOOKUP_SEP
)
hints = {"instance": self}
db_instance_qs = self.__class__._base_manager.db_manager(
using, hints=hints
).filter(pk=self.pk)
# Use provided fields, if not set then reload all non-deferred fields.
deferred_fields = self.get_deferred_fields()
if fields is not None:
fields = list(fields)
db_instance_qs = db_instance_qs.only(*fields)
elif deferred_fields:
fields = [
f.attname
for f in self._meta.concrete_fields
if f.attname not in deferred_fields
]
db_instance_qs = db_instance_qs.only(*fields)
db_instance = db_instance_qs.get()
non_loaded_fields = db_instance.get_deferred_fields()
for field in self._meta.concrete_fields:
if field.attname in non_loaded_fields:
# This field wasn't refreshed - skip ahead.
continue
setattr(self, field.attname, getattr(db_instance, field.attname))
# Clear cached foreign keys.
if field.is_relation and field.is_cached(self):
field.delete_cached_value(self)
# Clear cached relations.
for field in self._meta.related_objects:
if field.is_cached(self):
field.delete_cached_value(self)
self._state.db = db_instance._state.db
round¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
sample_size¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
save¶
Calls :meth:pre_save_polymorphic and saves the model.
View Source
save_base¶
def save_base(
self,
raw=False,
force_insert=False,
force_update=False,
using=None,
update_fields=None
)
Handle the parts of saving which should be done only once per save,
yet need to be done in raw saves, too. This includes some sanity checks and signal sending.
The 'raw' argument is telling save_base not to save any parent models and not to do any changes to the values before save. This is used by fixture loading.
View Source
def save_base(
self,
raw=False,
force_insert=False,
force_update=False,
using=None,
update_fields=None,
):
"""
Handle the parts of saving which should be done only once per save,
yet need to be done in raw saves, too. This includes some sanity
checks and signal sending.
The 'raw' argument is telling save_base not to save any parent
models and not to do any changes to the values before save. This
is used by fixture loading.
"""
using = using or router.db_for_write(self.__class__, instance=self)
assert not (force_insert and (force_update or update_fields))
assert update_fields is None or update_fields
cls = origin = self.__class__
# Skip proxies, but keep the origin as the proxy model.
if cls._meta.proxy:
cls = cls._meta.concrete_model
meta = cls._meta
if not meta.auto_created:
pre_save.send(
sender=origin,
instance=self,
raw=raw,
using=using,
update_fields=update_fields,
)
# A transaction isn't needed if one query is issued.
if meta.parents:
context_manager = transaction.atomic(using=using, savepoint=False)
else:
context_manager = transaction.mark_for_rollback_on_error(using=using)
with context_manager:
parent_inserted = False
if not raw:
parent_inserted = self._save_parents(cls, using, update_fields)
updated = self._save_table(
raw,
cls,
force_insert or parent_inserted,
force_update,
using,
update_fields,
)
# Store the database on which the object was saved
self._state.db = using
# Once saved, this is no longer a to-be-added instance.
self._state.adding = False
# Signal that the save is complete
if not meta.auto_created:
post_save.send(
sender=origin,
instance=self,
created=(not updated),
update_fields=update_fields,
raw=raw,
using=using,
)
serializable_value¶
Return the value of the field name for this instance. If the field is
a foreign key, return the id value instead of the object. If there's no Field object with this name on the model, return the model attribute's value.
Used to serialize a field's value (in the serializer, or form output, for example). Normally, you would just access the attribute directly and not use this method.
View Source
def serializable_value(self, field_name):
"""
Return the value of the field name for this instance. If the field is
a foreign key, return the id value instead of the object. If there's
no Field object with this name on the model, return the model
attribute's value.
Used to serialize a field's value (in the serializer, or form output,
for example). Normally, you would just access the attribute directly
and not use this method.
"""
try:
field = self._meta.get_field(field_name)
except FieldDoesNotExist:
return getattr(self, field_name)
return getattr(self, field.attname)
set_torch_model¶
Sets the model weights from a PyTorch model.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| value | torch.nn.Module | The PyTorch model. | None |
View Source
unique_error_message¶
View Source
def unique_error_message(self, model_class, unique_check):
opts = model_class._meta
params = {
"model": self,
"model_class": model_class,
"model_name": capfirst(opts.verbose_name),
"unique_check": unique_check,
}
# A unique field
if len(unique_check) == 1:
field = opts.get_field(unique_check[0])
params["field_label"] = capfirst(field.verbose_name)
return ValidationError(
message=field.error_messages["unique"],
code="unique",
params=params,
)
# unique_together
else:
field_labels = [
capfirst(opts.get_field(f).verbose_name) for f in unique_check
]
params["field_labels"] = get_text_list(field_labels, _("and"))
return ValidationError(
message=_("%(model_name)s with this %(field_labels)s already exists."),
code="unique_together",
params=params,
)
validate_unique¶
Check unique constraints on the model and raise ValidationError if any
failed.
View Source
def validate_unique(self, exclude=None):
"""
Check unique constraints on the model and raise ValidationError if any
failed.
"""
unique_checks, date_checks = self._get_unique_checks(exclude=exclude)
errors = self._perform_unique_checks(unique_checks)
date_errors = self._perform_date_checks(date_checks)
for k, v in date_errors.items():
errors.setdefault(k, []).extend(v)
if errors:
raise ValidationError(errors)
weights¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
MeanModel¶
Model class for mean models.
View Source
class MeanModel(GlobalModel):
"""
Model class for mean models.
"""
models: ManyToManyField = ManyToManyField(GlobalModel, related_name="mean_models")
"""Models of the mean model."""
def get_torch_model(self) -> torch.nn.Module:
"""
Converts the models to a PyTorch model.
Returns:
torch.nn.Module: The PyTorch model.
"""
torch_models: List[torch.nn.Module] = [model.get_torch_model() for model in self.models.all()]
model = MeanModule(torch_models)
if all(is_torchscript_instance(m) for m in torch_models):
return torch.jit.script(model)
return model
def set_torch_model(self, value: torch.nn.Module):
"""
Sets the models from a PyTorch model.
Args:
value (torch.nn.Module): The PyTorch model.
"""
raise NotImplementedError()
Ancestors (in MRO)¶
- fl_server_core.models.GlobalModel
- fl_server_core.models.Model
- polymorphic.models.PolymorphicModel
- django.db.models.base.Model
Class variables¶
Static methods¶
check¶
View Source
@classmethod
def check(cls, **kwargs):
errors = [
*cls._check_swappable(),
*cls._check_model(),
*cls._check_managers(**kwargs),
]
if not cls._meta.swapped:
databases = kwargs.get("databases") or []
errors += [
*cls._check_fields(**kwargs),
*cls._check_m2m_through_same_relationship(),
*cls._check_long_column_names(databases),
]
clash_errors = (
*cls._check_id_field(),
*cls._check_field_name_clashes(),
*cls._check_model_name_db_lookup_clashes(),
*cls._check_property_name_related_field_accessor_clashes(),
*cls._check_single_primary_key(),
)
errors.extend(clash_errors)
# If there are field name clashes, hide consequent column name
# clashes.
if not clash_errors:
errors.extend(cls._check_column_name_clashes())
errors += [
*cls._check_index_together(),
*cls._check_unique_together(),
*cls._check_indexes(databases),
*cls._check_ordering(),
*cls._check_constraints(databases),
*cls._check_default_pk(),
]
return errors
from_db¶
View Source
@classmethod
def from_db(cls, db, field_names, values):
if len(values) != len(cls._meta.concrete_fields):
values_iter = iter(values)
values = [
next(values_iter) if f.attname in field_names else DEFERRED
for f in cls._meta.concrete_fields
]
new = cls(*values)
new._state.adding = False
new._state.db = db
return new
translate_polymorphic_Q_object¶
View Source
Instance variables¶
Methods¶
clean¶
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.
View Source
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
clean_fields¶
Clean all fields and raise a ValidationError containing a dict
of all validation errors if any occur.
View Source
def clean_fields(self, exclude=None):
"""
Clean all fields and raise a ValidationError containing a dict
of all validation errors if any occur.
"""
if exclude is None:
exclude = []
errors = {}
for f in self._meta.fields:
if f.name in exclude:
continue
# Skip validation for empty fields with blank=True. The developer
# is responsible for making sure they have a valid value.
raw_value = getattr(self, f.attname)
if f.blank and raw_value in f.empty_values:
continue
try:
setattr(self, f.attname, f.clean(raw_value, self))
except ValidationError as e:
errors[f.name] = e.error_list
if errors:
raise ValidationError(errors)
date_error_message¶
View Source
def date_error_message(self, lookup_type, field_name, unique_for):
opts = self._meta
field = opts.get_field(field_name)
return ValidationError(
message=field.error_messages["unique_for_date"],
code="unique_for_date",
params={
"model": self,
"model_name": capfirst(opts.verbose_name),
"lookup_type": lookup_type,
"field": field_name,
"field_label": capfirst(field.verbose_name),
"date_field": unique_for,
"date_field_label": capfirst(opts.get_field(unique_for).verbose_name),
},
)
delete¶
View Source
def delete(self, using=None, keep_parents=False):
if self.pk is None:
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)
collector.collect([self], keep_parents=keep_parents)
return collector.delete()
description¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
full_clean¶
Call clean_fields(), clean(), and validate_unique() on the model.
Raise a ValidationError for any errors that occur.
View Source
def full_clean(self, exclude=None, validate_unique=True):
"""
Call clean_fields(), clean(), and validate_unique() on the model.
Raise a ValidationError for any errors that occur.
"""
errors = {}
if exclude is None:
exclude = []
else:
exclude = list(exclude)
try:
self.clean_fields(exclude=exclude)
except ValidationError as e:
errors = e.update_error_dict(errors)
# Form.clean() is run even if other validation fails, so do the
# same with Model.clean() for consistency.
try:
self.clean()
except ValidationError as e:
errors = e.update_error_dict(errors)
# Run unique checks, but only for fields that passed validation.
if validate_unique:
for name in errors:
if name != NON_FIELD_ERRORS and name not in exclude:
exclude.append(name)
try:
self.validate_unique(exclude=exclude)
except ValidationError as e:
errors = e.update_error_dict(errors)
if errors:
raise ValidationError(errors)
get_deferred_fields¶
Return a set containing names of deferred fields for this instance.
View Source
get_preprocessing_torch_model¶
Converts the preprocessing to a PyTorch model.
Returns:
| Type | Description |
|---|---|
| torch.nn.Module | The PyTorch model. |
View Source
get_real_concrete_instance_class¶
View Source
get_real_concrete_instance_class_id¶
View Source
get_real_instance¶
Upcast an object to it's actual type.
If a non-polymorphic manager (like base_objects) has been used to retrieve objects, then the complete object with it's real class/type and all fields may be retrieved with this method.
.. note::
Each method call executes one db query (if necessary).
Use the :meth:~polymorphic.managers.PolymorphicQuerySet.get_real_instances
to upcast a complete list in a single efficient query.
View Source
def get_real_instance(self):
"""
Upcast an object to it's actual type.
If a non-polymorphic manager (like base_objects) has been used to
retrieve objects, then the complete object with it's real class/type
and all fields may be retrieved with this method.
.. note::
Each method call executes one db query (if necessary).
Use the :meth:`~polymorphic.managers.PolymorphicQuerySet.get_real_instances`
to upcast a complete list in a single efficient query.
"""
real_model = self.get_real_instance_class()
if real_model == self.__class__:
return self
return real_model.objects.db_manager(self._state.db).get(pk=self.pk)
get_real_instance_class¶
Return the actual model type of the object.
If a non-polymorphic manager (like base_objects) has been used to retrieve objects, then the real class/type of these objects may be determined using this method.
View Source
def get_real_instance_class(self):
"""
Return the actual model type of the object.
If a non-polymorphic manager (like base_objects) has been used to
retrieve objects, then the real class/type of these objects may be
determined using this method.
"""
if self.polymorphic_ctype_id is None:
raise PolymorphicTypeUndefined(
(
"The model {}#{} does not have a `polymorphic_ctype_id` value defined.\n"
"If you created models outside polymorphic, e.g. through an import or migration, "
"make sure the `polymorphic_ctype_id` field points to the ContentType ID of the model subclass."
).format(self.__class__.__name__, self.pk)
)
# the following line would be the easiest way to do this, but it produces sql queries
# return self.polymorphic_ctype.model_class()
# so we use the following version, which uses the ContentType manager cache.
# Note that model_class() can return None for stale content types;
# when the content type record still exists but no longer refers to an existing model.
model = (
ContentType.objects.db_manager(self._state.db)
.get_for_id(self.polymorphic_ctype_id)
.model_class()
)
# Protect against bad imports (dumpdata without --natural) or other
# issues missing with the ContentType models.
if (
model is not None
and not issubclass(model, self.__class__)
and (
self.__class__._meta.proxy_for_model is None
or not issubclass(model, self.__class__._meta.proxy_for_model)
)
):
raise PolymorphicTypeInvalid(
"ContentType {} for {} #{} does not point to a subclass!".format(
self.polymorphic_ctype_id, model, self.pk
)
)
return model
get_torch_model¶
Converts the models to a PyTorch model.
Returns:
| Type | Description |
|---|---|
| torch.nn.Module | The PyTorch model. |
View Source
def get_torch_model(self) -> torch.nn.Module:
"""
Converts the models to a PyTorch model.
Returns:
torch.nn.Module: The PyTorch model.
"""
torch_models: List[torch.nn.Module] = [model.get_torch_model() for model in self.models.all()]
model = MeanModule(torch_models)
if all(is_torchscript_instance(m) for m in torch_models):
return torch.jit.script(model)
return model
get_training¶
Gets the training associated with the model.
Returns:
| Type | Description |
|---|---|
| models.Training | The training associated with the model. |
View Source
id¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
input_shape¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
is_global_model¶
Checks if the model is a global model.
Returns:
| Type | Description |
|---|---|
| bool | True if the model is a global model, False otherwise. |
View Source
is_local_model¶
Checks if the model is a local model.
Returns:
| Type | Description |
|---|---|
| bool | True if the model is a local model, False otherwise. |
View Source
name¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
pre_save_polymorphic¶
Make sure the polymorphic_ctype value is correctly set on this model.
View Source
def pre_save_polymorphic(self, using=DEFAULT_DB_ALIAS):
"""
Make sure the ``polymorphic_ctype`` value is correctly set on this model.
"""
# This function may be called manually in special use-cases. When the object
# is saved for the first time, we store its real class in polymorphic_ctype.
# When the object later is retrieved by PolymorphicQuerySet, it uses this
# field to figure out the real class of this object
# (used by PolymorphicQuerySet._get_real_instances)
if not self.polymorphic_ctype_id:
self.polymorphic_ctype = ContentType.objects.db_manager(using).get_for_model(
self, for_concrete_model=False
)
prepare_database_save¶
View Source
preprocessing¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
refresh_from_db¶
Reload field values from the database.
By default, the reloading happens from the database this instance was loaded from, or by the read router if this instance wasn't loaded from any database. The using parameter will override the default.
Fields can be used to specify which fields to reload. The fields should be an iterable of field attnames. If fields is None, then all non-deferred fields are reloaded.
When accessing deferred fields of an instance, the deferred loading of the field will call this method.
View Source
def refresh_from_db(self, using=None, fields=None):
"""
Reload field values from the database.
By default, the reloading happens from the database this instance was
loaded from, or by the read router if this instance wasn't loaded from
any database. The using parameter will override the default.
Fields can be used to specify which fields to reload. The fields
should be an iterable of field attnames. If fields is None, then
all non-deferred fields are reloaded.
When accessing deferred fields of an instance, the deferred loading
of the field will call this method.
"""
if fields is None:
self._prefetched_objects_cache = {}
else:
prefetched_objects_cache = getattr(self, "_prefetched_objects_cache", ())
for field in fields:
if field in prefetched_objects_cache:
del prefetched_objects_cache[field]
fields.remove(field)
if not fields:
return
if any(LOOKUP_SEP in f for f in fields):
raise ValueError(
'Found "%s" in fields argument. Relations and transforms '
"are not allowed in fields." % LOOKUP_SEP
)
hints = {"instance": self}
db_instance_qs = self.__class__._base_manager.db_manager(
using, hints=hints
).filter(pk=self.pk)
# Use provided fields, if not set then reload all non-deferred fields.
deferred_fields = self.get_deferred_fields()
if fields is not None:
fields = list(fields)
db_instance_qs = db_instance_qs.only(*fields)
elif deferred_fields:
fields = [
f.attname
for f in self._meta.concrete_fields
if f.attname not in deferred_fields
]
db_instance_qs = db_instance_qs.only(*fields)
db_instance = db_instance_qs.get()
non_loaded_fields = db_instance.get_deferred_fields()
for field in self._meta.concrete_fields:
if field.attname in non_loaded_fields:
# This field wasn't refreshed - skip ahead.
continue
setattr(self, field.attname, getattr(db_instance, field.attname))
# Clear cached foreign keys.
if field.is_relation and field.is_cached(self):
field.delete_cached_value(self)
# Clear cached relations.
for field in self._meta.related_objects:
if field.is_cached(self):
field.delete_cached_value(self)
self._state.db = db_instance._state.db
round¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
save¶
Calls :meth:pre_save_polymorphic and saves the model.
View Source
save_base¶
def save_base(
self,
raw=False,
force_insert=False,
force_update=False,
using=None,
update_fields=None
)
Handle the parts of saving which should be done only once per save,
yet need to be done in raw saves, too. This includes some sanity checks and signal sending.
The 'raw' argument is telling save_base not to save any parent models and not to do any changes to the values before save. This is used by fixture loading.
View Source
def save_base(
self,
raw=False,
force_insert=False,
force_update=False,
using=None,
update_fields=None,
):
"""
Handle the parts of saving which should be done only once per save,
yet need to be done in raw saves, too. This includes some sanity
checks and signal sending.
The 'raw' argument is telling save_base not to save any parent
models and not to do any changes to the values before save. This
is used by fixture loading.
"""
using = using or router.db_for_write(self.__class__, instance=self)
assert not (force_insert and (force_update or update_fields))
assert update_fields is None or update_fields
cls = origin = self.__class__
# Skip proxies, but keep the origin as the proxy model.
if cls._meta.proxy:
cls = cls._meta.concrete_model
meta = cls._meta
if not meta.auto_created:
pre_save.send(
sender=origin,
instance=self,
raw=raw,
using=using,
update_fields=update_fields,
)
# A transaction isn't needed if one query is issued.
if meta.parents:
context_manager = transaction.atomic(using=using, savepoint=False)
else:
context_manager = transaction.mark_for_rollback_on_error(using=using)
with context_manager:
parent_inserted = False
if not raw:
parent_inserted = self._save_parents(cls, using, update_fields)
updated = self._save_table(
raw,
cls,
force_insert or parent_inserted,
force_update,
using,
update_fields,
)
# Store the database on which the object was saved
self._state.db = using
# Once saved, this is no longer a to-be-added instance.
self._state.adding = False
# Signal that the save is complete
if not meta.auto_created:
post_save.send(
sender=origin,
instance=self,
created=(not updated),
update_fields=update_fields,
raw=raw,
using=using,
)
serializable_value¶
Return the value of the field name for this instance. If the field is
a foreign key, return the id value instead of the object. If there's no Field object with this name on the model, return the model attribute's value.
Used to serialize a field's value (in the serializer, or form output, for example). Normally, you would just access the attribute directly and not use this method.
View Source
def serializable_value(self, field_name):
"""
Return the value of the field name for this instance. If the field is
a foreign key, return the id value instead of the object. If there's
no Field object with this name on the model, return the model
attribute's value.
Used to serialize a field's value (in the serializer, or form output,
for example). Normally, you would just access the attribute directly
and not use this method.
"""
try:
field = self._meta.get_field(field_name)
except FieldDoesNotExist:
return getattr(self, field_name)
return getattr(self, field.attname)
set_preprocessing_torch_model¶
Sets the preprocessing from a PyTorch model.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| value | torch.nn.Module | The PyTorch model. | None |
View Source
set_torch_model¶
Sets the models from a PyTorch model.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| value | torch.nn.Module | The PyTorch model. | None |
View Source
unique_error_message¶
View Source
def unique_error_message(self, model_class, unique_check):
opts = model_class._meta
params = {
"model": self,
"model_class": model_class,
"model_name": capfirst(opts.verbose_name),
"unique_check": unique_check,
}
# A unique field
if len(unique_check) == 1:
field = opts.get_field(unique_check[0])
params["field_label"] = capfirst(field.verbose_name)
return ValidationError(
message=field.error_messages["unique"],
code="unique",
params=params,
)
# unique_together
else:
field_labels = [
capfirst(opts.get_field(f).verbose_name) for f in unique_check
]
params["field_labels"] = get_text_list(field_labels, _("and"))
return ValidationError(
message=_("%(model_name)s with this %(field_labels)s already exists."),
code="unique_together",
params=params,
)
validate_unique¶
Check unique constraints on the model and raise ValidationError if any
failed.
View Source
def validate_unique(self, exclude=None):
"""
Check unique constraints on the model and raise ValidationError if any
failed.
"""
unique_checks, date_checks = self._get_unique_checks(exclude=exclude)
errors = self._perform_unique_checks(unique_checks)
date_errors = self._perform_date_checks(date_checks)
for k, v in date_errors.items():
errors.setdefault(k, []).extend(v)
if errors:
raise ValidationError(errors)
weights¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
Metric¶
Metric model class.
View Source
class Metric(models.Model):
"""
Metric model class.
"""
model: ForeignKey = ForeignKey(Model, on_delete=CASCADE)
"""Model associated with the metric."""
identifier: CharField = CharField(max_length=64, null=True, blank=True)
"""Identifier of the metric."""
key: CharField = CharField(max_length=32)
"""Key of the metric."""
value_float: FloatField = FloatField(null=True, blank=True)
"""Float value of the metric."""
value_binary: BinaryField = BinaryField(null=True, blank=True)
"""Binary value of the metric."""
step: IntegerField = IntegerField(null=True, blank=True)
"""Step of the metric."""
reporter: ForeignKey = ForeignKey(User, null=True, blank=True, on_delete=CASCADE)
"""User who reported the metric."""
@property
def value(self) -> float | bytes:
"""
Value of the metric.
Returns:
float | bytes: The value of the metric.
"""
if self.is_float():
return self.value_float
return self.value_binary
@value.setter
def value(self, value: float | int | bytes | Module | Tensor):
"""
Setter for the value of the metric.
Args:
value (float | int | bytes | Module | Tensor): The value to set.
"""
if isinstance(value, float):
self.value_float = value
elif isinstance(value, int):
self.value_float = float(value)
elif isinstance(value, (Module, Tensor)):
self.value_binary = from_torch_module_or_tensor(value)
else:
self.value_binary = value
@value.deleter
def value(self):
"""
Deleter for the value of the metric.
"""
self.value_float = None
self.value_binary = None
def is_float(self) -> bool:
"""
Check if the value of the metric is a float.
Returns:
bool: `True` if the value of the metric is a float, otherwise `False`.
"""
return self.value_float is not None
def is_binary(self) -> bool:
"""
Check if the value of the metric is binary.
Returns:
bool: `True` if the value of the metric is binary, otherwise `False`.
"""
return self.value_binary is not None
def to_torch(self) -> Module | Tensor:
"""
Convert the binary value of the metric to a torch module or tensor.
Returns:
Module | Tensor: The converted torch module or tensor.
"""
return to_torch_module_or_tensor(self.value_binary)
Ancestors (in MRO)¶
- django.db.models.base.Model
Class variables¶
Static methods¶
check¶
View Source
@classmethod
def check(cls, **kwargs):
errors = [
*cls._check_swappable(),
*cls._check_model(),
*cls._check_managers(**kwargs),
]
if not cls._meta.swapped:
databases = kwargs.get("databases") or []
errors += [
*cls._check_fields(**kwargs),
*cls._check_m2m_through_same_relationship(),
*cls._check_long_column_names(databases),
]
clash_errors = (
*cls._check_id_field(),
*cls._check_field_name_clashes(),
*cls._check_model_name_db_lookup_clashes(),
*cls._check_property_name_related_field_accessor_clashes(),
*cls._check_single_primary_key(),
)
errors.extend(clash_errors)
# If there are field name clashes, hide consequent column name
# clashes.
if not clash_errors:
errors.extend(cls._check_column_name_clashes())
errors += [
*cls._check_index_together(),
*cls._check_unique_together(),
*cls._check_indexes(databases),
*cls._check_ordering(),
*cls._check_constraints(databases),
*cls._check_default_pk(),
]
return errors
from_db¶
View Source
@classmethod
def from_db(cls, db, field_names, values):
if len(values) != len(cls._meta.concrete_fields):
values_iter = iter(values)
values = [
next(values_iter) if f.attname in field_names else DEFERRED
for f in cls._meta.concrete_fields
]
new = cls(*values)
new._state.adding = False
new._state.db = db
return new
Instance variables¶
Value of the metric.
Methods¶
clean¶
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.
View Source
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
clean_fields¶
Clean all fields and raise a ValidationError containing a dict
of all validation errors if any occur.
View Source
def clean_fields(self, exclude=None):
"""
Clean all fields and raise a ValidationError containing a dict
of all validation errors if any occur.
"""
if exclude is None:
exclude = []
errors = {}
for f in self._meta.fields:
if f.name in exclude:
continue
# Skip validation for empty fields with blank=True. The developer
# is responsible for making sure they have a valid value.
raw_value = getattr(self, f.attname)
if f.blank and raw_value in f.empty_values:
continue
try:
setattr(self, f.attname, f.clean(raw_value, self))
except ValidationError as e:
errors[f.name] = e.error_list
if errors:
raise ValidationError(errors)
date_error_message¶
View Source
def date_error_message(self, lookup_type, field_name, unique_for):
opts = self._meta
field = opts.get_field(field_name)
return ValidationError(
message=field.error_messages["unique_for_date"],
code="unique_for_date",
params={
"model": self,
"model_name": capfirst(opts.verbose_name),
"lookup_type": lookup_type,
"field": field_name,
"field_label": capfirst(field.verbose_name),
"date_field": unique_for,
"date_field_label": capfirst(opts.get_field(unique_for).verbose_name),
},
)
delete¶
View Source
def delete(self, using=None, keep_parents=False):
if self.pk is None:
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)
collector.collect([self], keep_parents=keep_parents)
return collector.delete()
full_clean¶
Call clean_fields(), clean(), and validate_unique() on the model.
Raise a ValidationError for any errors that occur.
View Source
def full_clean(self, exclude=None, validate_unique=True):
"""
Call clean_fields(), clean(), and validate_unique() on the model.
Raise a ValidationError for any errors that occur.
"""
errors = {}
if exclude is None:
exclude = []
else:
exclude = list(exclude)
try:
self.clean_fields(exclude=exclude)
except ValidationError as e:
errors = e.update_error_dict(errors)
# Form.clean() is run even if other validation fails, so do the
# same with Model.clean() for consistency.
try:
self.clean()
except ValidationError as e:
errors = e.update_error_dict(errors)
# Run unique checks, but only for fields that passed validation.
if validate_unique:
for name in errors:
if name != NON_FIELD_ERRORS and name not in exclude:
exclude.append(name)
try:
self.validate_unique(exclude=exclude)
except ValidationError as e:
errors = e.update_error_dict(errors)
if errors:
raise ValidationError(errors)
get_deferred_fields¶
Return a set containing names of deferred fields for this instance.
View Source
id¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
identifier¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
is_binary¶
Check if the value of the metric is binary.
Returns:
| Type | Description |
|---|---|
| bool | True if the value of the metric is binary, otherwise False. |
View Source
is_float¶
Check if the value of the metric is a float.
Returns:
| Type | Description |
|---|---|
| bool | True if the value of the metric is a float, otherwise False. |
View Source
key¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
prepare_database_save¶
View Source
refresh_from_db¶
Reload field values from the database.
By default, the reloading happens from the database this instance was loaded from, or by the read router if this instance wasn't loaded from any database. The using parameter will override the default.
Fields can be used to specify which fields to reload. The fields should be an iterable of field attnames. If fields is None, then all non-deferred fields are reloaded.
When accessing deferred fields of an instance, the deferred loading of the field will call this method.
View Source
def refresh_from_db(self, using=None, fields=None):
"""
Reload field values from the database.
By default, the reloading happens from the database this instance was
loaded from, or by the read router if this instance wasn't loaded from
any database. The using parameter will override the default.
Fields can be used to specify which fields to reload. The fields
should be an iterable of field attnames. If fields is None, then
all non-deferred fields are reloaded.
When accessing deferred fields of an instance, the deferred loading
of the field will call this method.
"""
if fields is None:
self._prefetched_objects_cache = {}
else:
prefetched_objects_cache = getattr(self, "_prefetched_objects_cache", ())
for field in fields:
if field in prefetched_objects_cache:
del prefetched_objects_cache[field]
fields.remove(field)
if not fields:
return
if any(LOOKUP_SEP in f for f in fields):
raise ValueError(
'Found "%s" in fields argument. Relations and transforms '
"are not allowed in fields." % LOOKUP_SEP
)
hints = {"instance": self}
db_instance_qs = self.__class__._base_manager.db_manager(
using, hints=hints
).filter(pk=self.pk)
# Use provided fields, if not set then reload all non-deferred fields.
deferred_fields = self.get_deferred_fields()
if fields is not None:
fields = list(fields)
db_instance_qs = db_instance_qs.only(*fields)
elif deferred_fields:
fields = [
f.attname
for f in self._meta.concrete_fields
if f.attname not in deferred_fields
]
db_instance_qs = db_instance_qs.only(*fields)
db_instance = db_instance_qs.get()
non_loaded_fields = db_instance.get_deferred_fields()
for field in self._meta.concrete_fields:
if field.attname in non_loaded_fields:
# This field wasn't refreshed - skip ahead.
continue
setattr(self, field.attname, getattr(db_instance, field.attname))
# Clear cached foreign keys.
if field.is_relation and field.is_cached(self):
field.delete_cached_value(self)
# Clear cached relations.
for field in self._meta.related_objects:
if field.is_cached(self):
field.delete_cached_value(self)
self._state.db = db_instance._state.db
save¶
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.
View Source
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_fields = self.get_deferred_fields()
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 = set()
for field in self._meta.concrete_fields:
if not field.primary_key:
field_names.add(field.name)
if field.name != field.attname:
field_names.add(field.attname)
non_model_fields = update_fields.difference(field_names)
if non_model_fields:
raise ValueError(
"The following fields do not exist in this model, are m2m "
"fields, or are non-concrete fields: %s"
% ", ".join(non_model_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_fields and using == self._state.db:
field_names = set()
for field in self._meta.concrete_fields:
if not field.primary_key and not hasattr(field, "through"):
field_names.add(field.attname)
loaded_fields = field_names.difference(deferred_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,
)
save_base¶
def save_base(
self,
raw=False,
force_insert=False,
force_update=False,
using=None,
update_fields=None
)
Handle the parts of saving which should be done only once per save,
yet need to be done in raw saves, too. This includes some sanity checks and signal sending.
The 'raw' argument is telling save_base not to save any parent models and not to do any changes to the values before save. This is used by fixture loading.
View Source
def save_base(
self,
raw=False,
force_insert=False,
force_update=False,
using=None,
update_fields=None,
):
"""
Handle the parts of saving which should be done only once per save,
yet need to be done in raw saves, too. This includes some sanity
checks and signal sending.
The 'raw' argument is telling save_base not to save any parent
models and not to do any changes to the values before save. This
is used by fixture loading.
"""
using = using or router.db_for_write(self.__class__, instance=self)
assert not (force_insert and (force_update or update_fields))
assert update_fields is None or update_fields
cls = origin = self.__class__
# Skip proxies, but keep the origin as the proxy model.
if cls._meta.proxy:
cls = cls._meta.concrete_model
meta = cls._meta
if not meta.auto_created:
pre_save.send(
sender=origin,
instance=self,
raw=raw,
using=using,
update_fields=update_fields,
)
# A transaction isn't needed if one query is issued.
if meta.parents:
context_manager = transaction.atomic(using=using, savepoint=False)
else:
context_manager = transaction.mark_for_rollback_on_error(using=using)
with context_manager:
parent_inserted = False
if not raw:
parent_inserted = self._save_parents(cls, using, update_fields)
updated = self._save_table(
raw,
cls,
force_insert or parent_inserted,
force_update,
using,
update_fields,
)
# Store the database on which the object was saved
self._state.db = using
# Once saved, this is no longer a to-be-added instance.
self._state.adding = False
# Signal that the save is complete
if not meta.auto_created:
post_save.send(
sender=origin,
instance=self,
created=(not updated),
update_fields=update_fields,
raw=raw,
using=using,
)
serializable_value¶
Return the value of the field name for this instance. If the field is
a foreign key, return the id value instead of the object. If there's no Field object with this name on the model, return the model attribute's value.
Used to serialize a field's value (in the serializer, or form output, for example). Normally, you would just access the attribute directly and not use this method.
View Source
def serializable_value(self, field_name):
"""
Return the value of the field name for this instance. If the field is
a foreign key, return the id value instead of the object. If there's
no Field object with this name on the model, return the model
attribute's value.
Used to serialize a field's value (in the serializer, or form output,
for example). Normally, you would just access the attribute directly
and not use this method.
"""
try:
field = self._meta.get_field(field_name)
except FieldDoesNotExist:
return getattr(self, field_name)
return getattr(self, field.attname)
step¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
to_torch¶
Convert the binary value of the metric to a torch module or tensor.
Returns:
| Type | Description |
|---|---|
| None | Module |
View Source
unique_error_message¶
View Source
def unique_error_message(self, model_class, unique_check):
opts = model_class._meta
params = {
"model": self,
"model_class": model_class,
"model_name": capfirst(opts.verbose_name),
"unique_check": unique_check,
}
# A unique field
if len(unique_check) == 1:
field = opts.get_field(unique_check[0])
params["field_label"] = capfirst(field.verbose_name)
return ValidationError(
message=field.error_messages["unique"],
code="unique",
params=params,
)
# unique_together
else:
field_labels = [
capfirst(opts.get_field(f).verbose_name) for f in unique_check
]
params["field_labels"] = get_text_list(field_labels, _("and"))
return ValidationError(
message=_("%(model_name)s with this %(field_labels)s already exists."),
code="unique_together",
params=params,
)
validate_unique¶
Check unique constraints on the model and raise ValidationError if any
failed.
View Source
def validate_unique(self, exclude=None):
"""
Check unique constraints on the model and raise ValidationError if any
failed.
"""
unique_checks, date_checks = self._get_unique_checks(exclude=exclude)
errors = self._perform_unique_checks(unique_checks)
date_errors = self._perform_date_checks(date_checks)
for k, v in date_errors.items():
errors.setdefault(k, []).extend(v)
if errors:
raise ValidationError(errors)
value_binary¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
value_float¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
Model¶
Base model class for all types of model models.
View Source
class Model(PolymorphicModel):
"""
Base model class for all types of model models.
"""
id: UUIDField = UUIDField(primary_key=True, editable=False, default=uuid4)
"""Unique identifier for the model."""
owner: ForeignKey = ForeignKey(User, on_delete=CASCADE)
"""User who owns the model."""
round: IntegerField = IntegerField()
"""Round number of the model."""
weights: BinaryField = BinaryField()
"""Weights of the model."""
def is_global_model(self):
"""
Checks if the model is a global model.
Returns:
bool: True if the model is a global model, False otherwise.
"""
return isinstance(self, GlobalModel)
def is_local_model(self):
"""
Checks if the model is a local model.
Returns:
bool: True if the model is a local model, False otherwise.
"""
return isinstance(self, LocalModel)
def get_torch_model(self) -> torch.nn.Module:
"""
Converts the model weights to a PyTorch model.
Returns:
torch.nn.Module: The PyTorch model.
"""
return to_torch_module(self.weights)
def set_torch_model(self, value: torch.nn.Module):
"""
Sets the model weights from a PyTorch model.
Args:
value (torch.nn.Module): The PyTorch model.
"""
self.weights = from_torch_module(value)
def get_training(self) -> Optional["models.Training"]:
"""
Gets the training associated with the model.
Returns:
models.Training: The training associated with the model.
"""
return models.Training.objects.filter(model=self).first()
Ancestors (in MRO)¶
- polymorphic.models.PolymorphicModel
- django.db.models.base.Model
Descendants¶
- fl_server_core.models.GlobalModel
- fl_server_core.models.LocalModel
Class variables¶
Static methods¶
check¶
View Source
@classmethod
def check(cls, **kwargs):
errors = [
*cls._check_swappable(),
*cls._check_model(),
*cls._check_managers(**kwargs),
]
if not cls._meta.swapped:
databases = kwargs.get("databases") or []
errors += [
*cls._check_fields(**kwargs),
*cls._check_m2m_through_same_relationship(),
*cls._check_long_column_names(databases),
]
clash_errors = (
*cls._check_id_field(),
*cls._check_field_name_clashes(),
*cls._check_model_name_db_lookup_clashes(),
*cls._check_property_name_related_field_accessor_clashes(),
*cls._check_single_primary_key(),
)
errors.extend(clash_errors)
# If there are field name clashes, hide consequent column name
# clashes.
if not clash_errors:
errors.extend(cls._check_column_name_clashes())
errors += [
*cls._check_index_together(),
*cls._check_unique_together(),
*cls._check_indexes(databases),
*cls._check_ordering(),
*cls._check_constraints(databases),
*cls._check_default_pk(),
]
return errors
from_db¶
View Source
@classmethod
def from_db(cls, db, field_names, values):
if len(values) != len(cls._meta.concrete_fields):
values_iter = iter(values)
values = [
next(values_iter) if f.attname in field_names else DEFERRED
for f in cls._meta.concrete_fields
]
new = cls(*values)
new._state.adding = False
new._state.db = db
return new
translate_polymorphic_Q_object¶
View Source
Instance variables¶
Methods¶
clean¶
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.
View Source
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
clean_fields¶
Clean all fields and raise a ValidationError containing a dict
of all validation errors if any occur.
View Source
def clean_fields(self, exclude=None):
"""
Clean all fields and raise a ValidationError containing a dict
of all validation errors if any occur.
"""
if exclude is None:
exclude = []
errors = {}
for f in self._meta.fields:
if f.name in exclude:
continue
# Skip validation for empty fields with blank=True. The developer
# is responsible for making sure they have a valid value.
raw_value = getattr(self, f.attname)
if f.blank and raw_value in f.empty_values:
continue
try:
setattr(self, f.attname, f.clean(raw_value, self))
except ValidationError as e:
errors[f.name] = e.error_list
if errors:
raise ValidationError(errors)
date_error_message¶
View Source
def date_error_message(self, lookup_type, field_name, unique_for):
opts = self._meta
field = opts.get_field(field_name)
return ValidationError(
message=field.error_messages["unique_for_date"],
code="unique_for_date",
params={
"model": self,
"model_name": capfirst(opts.verbose_name),
"lookup_type": lookup_type,
"field": field_name,
"field_label": capfirst(field.verbose_name),
"date_field": unique_for,
"date_field_label": capfirst(opts.get_field(unique_for).verbose_name),
},
)
delete¶
View Source
def delete(self, using=None, keep_parents=False):
if self.pk is None:
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)
collector.collect([self], keep_parents=keep_parents)
return collector.delete()
full_clean¶
Call clean_fields(), clean(), and validate_unique() on the model.
Raise a ValidationError for any errors that occur.
View Source
def full_clean(self, exclude=None, validate_unique=True):
"""
Call clean_fields(), clean(), and validate_unique() on the model.
Raise a ValidationError for any errors that occur.
"""
errors = {}
if exclude is None:
exclude = []
else:
exclude = list(exclude)
try:
self.clean_fields(exclude=exclude)
except ValidationError as e:
errors = e.update_error_dict(errors)
# Form.clean() is run even if other validation fails, so do the
# same with Model.clean() for consistency.
try:
self.clean()
except ValidationError as e:
errors = e.update_error_dict(errors)
# Run unique checks, but only for fields that passed validation.
if validate_unique:
for name in errors:
if name != NON_FIELD_ERRORS and name not in exclude:
exclude.append(name)
try:
self.validate_unique(exclude=exclude)
except ValidationError as e:
errors = e.update_error_dict(errors)
if errors:
raise ValidationError(errors)
get_deferred_fields¶
Return a set containing names of deferred fields for this instance.
View Source
get_real_concrete_instance_class¶
View Source
get_real_concrete_instance_class_id¶
View Source
get_real_instance¶
Upcast an object to it's actual type.
If a non-polymorphic manager (like base_objects) has been used to retrieve objects, then the complete object with it's real class/type and all fields may be retrieved with this method.
.. note::
Each method call executes one db query (if necessary).
Use the :meth:~polymorphic.managers.PolymorphicQuerySet.get_real_instances
to upcast a complete list in a single efficient query.
View Source
def get_real_instance(self):
"""
Upcast an object to it's actual type.
If a non-polymorphic manager (like base_objects) has been used to
retrieve objects, then the complete object with it's real class/type
and all fields may be retrieved with this method.
.. note::
Each method call executes one db query (if necessary).
Use the :meth:`~polymorphic.managers.PolymorphicQuerySet.get_real_instances`
to upcast a complete list in a single efficient query.
"""
real_model = self.get_real_instance_class()
if real_model == self.__class__:
return self
return real_model.objects.db_manager(self._state.db).get(pk=self.pk)
get_real_instance_class¶
Return the actual model type of the object.
If a non-polymorphic manager (like base_objects) has been used to retrieve objects, then the real class/type of these objects may be determined using this method.
View Source
def get_real_instance_class(self):
"""
Return the actual model type of the object.
If a non-polymorphic manager (like base_objects) has been used to
retrieve objects, then the real class/type of these objects may be
determined using this method.
"""
if self.polymorphic_ctype_id is None:
raise PolymorphicTypeUndefined(
(
"The model {}#{} does not have a `polymorphic_ctype_id` value defined.\n"
"If you created models outside polymorphic, e.g. through an import or migration, "
"make sure the `polymorphic_ctype_id` field points to the ContentType ID of the model subclass."
).format(self.__class__.__name__, self.pk)
)
# the following line would be the easiest way to do this, but it produces sql queries
# return self.polymorphic_ctype.model_class()
# so we use the following version, which uses the ContentType manager cache.
# Note that model_class() can return None for stale content types;
# when the content type record still exists but no longer refers to an existing model.
model = (
ContentType.objects.db_manager(self._state.db)
.get_for_id(self.polymorphic_ctype_id)
.model_class()
)
# Protect against bad imports (dumpdata without --natural) or other
# issues missing with the ContentType models.
if (
model is not None
and not issubclass(model, self.__class__)
and (
self.__class__._meta.proxy_for_model is None
or not issubclass(model, self.__class__._meta.proxy_for_model)
)
):
raise PolymorphicTypeInvalid(
"ContentType {} for {} #{} does not point to a subclass!".format(
self.polymorphic_ctype_id, model, self.pk
)
)
return model
get_torch_model¶
Converts the model weights to a PyTorch model.
Returns:
| Type | Description |
|---|---|
| torch.nn.Module | The PyTorch model. |
View Source
get_training¶
Gets the training associated with the model.
Returns:
| Type | Description |
|---|---|
| models.Training | The training associated with the model. |
View Source
id¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
is_global_model¶
Checks if the model is a global model.
Returns:
| Type | Description |
|---|---|
| bool | True if the model is a global model, False otherwise. |
View Source
is_local_model¶
Checks if the model is a local model.
Returns:
| Type | Description |
|---|---|
| bool | True if the model is a local model, False otherwise. |
View Source
pre_save_polymorphic¶
Make sure the polymorphic_ctype value is correctly set on this model.
View Source
def pre_save_polymorphic(self, using=DEFAULT_DB_ALIAS):
"""
Make sure the ``polymorphic_ctype`` value is correctly set on this model.
"""
# This function may be called manually in special use-cases. When the object
# is saved for the first time, we store its real class in polymorphic_ctype.
# When the object later is retrieved by PolymorphicQuerySet, it uses this
# field to figure out the real class of this object
# (used by PolymorphicQuerySet._get_real_instances)
if not self.polymorphic_ctype_id:
self.polymorphic_ctype = ContentType.objects.db_manager(using).get_for_model(
self, for_concrete_model=False
)
prepare_database_save¶
View Source
refresh_from_db¶
Reload field values from the database.
By default, the reloading happens from the database this instance was loaded from, or by the read router if this instance wasn't loaded from any database. The using parameter will override the default.
Fields can be used to specify which fields to reload. The fields should be an iterable of field attnames. If fields is None, then all non-deferred fields are reloaded.
When accessing deferred fields of an instance, the deferred loading of the field will call this method.
View Source
def refresh_from_db(self, using=None, fields=None):
"""
Reload field values from the database.
By default, the reloading happens from the database this instance was
loaded from, or by the read router if this instance wasn't loaded from
any database. The using parameter will override the default.
Fields can be used to specify which fields to reload. The fields
should be an iterable of field attnames. If fields is None, then
all non-deferred fields are reloaded.
When accessing deferred fields of an instance, the deferred loading
of the field will call this method.
"""
if fields is None:
self._prefetched_objects_cache = {}
else:
prefetched_objects_cache = getattr(self, "_prefetched_objects_cache", ())
for field in fields:
if field in prefetched_objects_cache:
del prefetched_objects_cache[field]
fields.remove(field)
if not fields:
return
if any(LOOKUP_SEP in f for f in fields):
raise ValueError(
'Found "%s" in fields argument. Relations and transforms '
"are not allowed in fields." % LOOKUP_SEP
)
hints = {"instance": self}
db_instance_qs = self.__class__._base_manager.db_manager(
using, hints=hints
).filter(pk=self.pk)
# Use provided fields, if not set then reload all non-deferred fields.
deferred_fields = self.get_deferred_fields()
if fields is not None:
fields = list(fields)
db_instance_qs = db_instance_qs.only(*fields)
elif deferred_fields:
fields = [
f.attname
for f in self._meta.concrete_fields
if f.attname not in deferred_fields
]
db_instance_qs = db_instance_qs.only(*fields)
db_instance = db_instance_qs.get()
non_loaded_fields = db_instance.get_deferred_fields()
for field in self._meta.concrete_fields:
if field.attname in non_loaded_fields:
# This field wasn't refreshed - skip ahead.
continue
setattr(self, field.attname, getattr(db_instance, field.attname))
# Clear cached foreign keys.
if field.is_relation and field.is_cached(self):
field.delete_cached_value(self)
# Clear cached relations.
for field in self._meta.related_objects:
if field.is_cached(self):
field.delete_cached_value(self)
self._state.db = db_instance._state.db
round¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
save¶
Calls :meth:pre_save_polymorphic and saves the model.
View Source
save_base¶
def save_base(
self,
raw=False,
force_insert=False,
force_update=False,
using=None,
update_fields=None
)
Handle the parts of saving which should be done only once per save,
yet need to be done in raw saves, too. This includes some sanity checks and signal sending.
The 'raw' argument is telling save_base not to save any parent models and not to do any changes to the values before save. This is used by fixture loading.
View Source
def save_base(
self,
raw=False,
force_insert=False,
force_update=False,
using=None,
update_fields=None,
):
"""
Handle the parts of saving which should be done only once per save,
yet need to be done in raw saves, too. This includes some sanity
checks and signal sending.
The 'raw' argument is telling save_base not to save any parent
models and not to do any changes to the values before save. This
is used by fixture loading.
"""
using = using or router.db_for_write(self.__class__, instance=self)
assert not (force_insert and (force_update or update_fields))
assert update_fields is None or update_fields
cls = origin = self.__class__
# Skip proxies, but keep the origin as the proxy model.
if cls._meta.proxy:
cls = cls._meta.concrete_model
meta = cls._meta
if not meta.auto_created:
pre_save.send(
sender=origin,
instance=self,
raw=raw,
using=using,
update_fields=update_fields,
)
# A transaction isn't needed if one query is issued.
if meta.parents:
context_manager = transaction.atomic(using=using, savepoint=False)
else:
context_manager = transaction.mark_for_rollback_on_error(using=using)
with context_manager:
parent_inserted = False
if not raw:
parent_inserted = self._save_parents(cls, using, update_fields)
updated = self._save_table(
raw,
cls,
force_insert or parent_inserted,
force_update,
using,
update_fields,
)
# Store the database on which the object was saved
self._state.db = using
# Once saved, this is no longer a to-be-added instance.
self._state.adding = False
# Signal that the save is complete
if not meta.auto_created:
post_save.send(
sender=origin,
instance=self,
created=(not updated),
update_fields=update_fields,
raw=raw,
using=using,
)
serializable_value¶
Return the value of the field name for this instance. If the field is
a foreign key, return the id value instead of the object. If there's no Field object with this name on the model, return the model attribute's value.
Used to serialize a field's value (in the serializer, or form output, for example). Normally, you would just access the attribute directly and not use this method.
View Source
def serializable_value(self, field_name):
"""
Return the value of the field name for this instance. If the field is
a foreign key, return the id value instead of the object. If there's
no Field object with this name on the model, return the model
attribute's value.
Used to serialize a field's value (in the serializer, or form output,
for example). Normally, you would just access the attribute directly
and not use this method.
"""
try:
field = self._meta.get_field(field_name)
except FieldDoesNotExist:
return getattr(self, field_name)
return getattr(self, field.attname)
set_torch_model¶
Sets the model weights from a PyTorch model.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| value | torch.nn.Module | The PyTorch model. | None |
View Source
unique_error_message¶
View Source
def unique_error_message(self, model_class, unique_check):
opts = model_class._meta
params = {
"model": self,
"model_class": model_class,
"model_name": capfirst(opts.verbose_name),
"unique_check": unique_check,
}
# A unique field
if len(unique_check) == 1:
field = opts.get_field(unique_check[0])
params["field_label"] = capfirst(field.verbose_name)
return ValidationError(
message=field.error_messages["unique"],
code="unique",
params=params,
)
# unique_together
else:
field_labels = [
capfirst(opts.get_field(f).verbose_name) for f in unique_check
]
params["field_labels"] = get_text_list(field_labels, _("and"))
return ValidationError(
message=_("%(model_name)s with this %(field_labels)s already exists."),
code="unique_together",
params=params,
)
validate_unique¶
Check unique constraints on the model and raise ValidationError if any
failed.
View Source
def validate_unique(self, exclude=None):
"""
Check unique constraints on the model and raise ValidationError if any
failed.
"""
unique_checks, date_checks = self._get_unique_checks(exclude=exclude)
errors = self._perform_unique_checks(unique_checks)
date_errors = self._perform_date_checks(date_checks)
for k, v in date_errors.items():
errors.setdefault(k, []).extend(v)
if errors:
raise ValidationError(errors)
weights¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
SWAGModel¶
Model class for SWAG models.
View Source
class SWAGModel(GlobalModel):
"""
Model class for SWAG models.
"""
swag_first_moment: BinaryField = BinaryField()
"""First moment of the SWAG model."""
swag_second_moment: BinaryField = BinaryField()
"""Second moment of the SWAG model."""
@property
def first_moment(self) -> Tensor:
"""
Gets the first moment of the SWAG model.
Returns:
Tensor: The first moment of the SWAG model.
"""
return to_torch_tensor(self.swag_first_moment)
@first_moment.setter
def first_moment(self, value: Tensor):
"""
Sets the first moment of the SWAG model.
Args:
value (Tensor): The first moment of the SWAG model.
"""
self.swag_first_moment = from_torch_tensor(value)
@property
def second_moment(self) -> Tensor:
"""
Gets the second moment of the SWAG model.
Returns:
Tensor: The second moment of the SWAG model.
"""
return to_torch_tensor(self.swag_second_moment)
@second_moment.setter
def second_moment(self, value: Tensor):
"""
Sets the second moment of the SWAG model.
Args:
value (Tensor): The second moment of the SWAG model.
"""
self.swag_second_moment = from_torch_tensor(value)
Ancestors (in MRO)¶
- fl_server_core.models.GlobalModel
- fl_server_core.models.Model
- polymorphic.models.PolymorphicModel
- django.db.models.base.Model
Class variables¶
Static methods¶
check¶
View Source
@classmethod
def check(cls, **kwargs):
errors = [
*cls._check_swappable(),
*cls._check_model(),
*cls._check_managers(**kwargs),
]
if not cls._meta.swapped:
databases = kwargs.get("databases") or []
errors += [
*cls._check_fields(**kwargs),
*cls._check_m2m_through_same_relationship(),
*cls._check_long_column_names(databases),
]
clash_errors = (
*cls._check_id_field(),
*cls._check_field_name_clashes(),
*cls._check_model_name_db_lookup_clashes(),
*cls._check_property_name_related_field_accessor_clashes(),
*cls._check_single_primary_key(),
)
errors.extend(clash_errors)
# If there are field name clashes, hide consequent column name
# clashes.
if not clash_errors:
errors.extend(cls._check_column_name_clashes())
errors += [
*cls._check_index_together(),
*cls._check_unique_together(),
*cls._check_indexes(databases),
*cls._check_ordering(),
*cls._check_constraints(databases),
*cls._check_default_pk(),
]
return errors
from_db¶
View Source
@classmethod
def from_db(cls, db, field_names, values):
if len(values) != len(cls._meta.concrete_fields):
values_iter = iter(values)
values = [
next(values_iter) if f.attname in field_names else DEFERRED
for f in cls._meta.concrete_fields
]
new = cls(*values)
new._state.adding = False
new._state.db = db
return new
translate_polymorphic_Q_object¶
View Source
Instance variables¶
Gets the first moment of the SWAG model.
Gets the second moment of the SWAG model.
Methods¶
clean¶
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.
View Source
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
clean_fields¶
Clean all fields and raise a ValidationError containing a dict
of all validation errors if any occur.
View Source
def clean_fields(self, exclude=None):
"""
Clean all fields and raise a ValidationError containing a dict
of all validation errors if any occur.
"""
if exclude is None:
exclude = []
errors = {}
for f in self._meta.fields:
if f.name in exclude:
continue
# Skip validation for empty fields with blank=True. The developer
# is responsible for making sure they have a valid value.
raw_value = getattr(self, f.attname)
if f.blank and raw_value in f.empty_values:
continue
try:
setattr(self, f.attname, f.clean(raw_value, self))
except ValidationError as e:
errors[f.name] = e.error_list
if errors:
raise ValidationError(errors)
date_error_message¶
View Source
def date_error_message(self, lookup_type, field_name, unique_for):
opts = self._meta
field = opts.get_field(field_name)
return ValidationError(
message=field.error_messages["unique_for_date"],
code="unique_for_date",
params={
"model": self,
"model_name": capfirst(opts.verbose_name),
"lookup_type": lookup_type,
"field": field_name,
"field_label": capfirst(field.verbose_name),
"date_field": unique_for,
"date_field_label": capfirst(opts.get_field(unique_for).verbose_name),
},
)
delete¶
View Source
def delete(self, using=None, keep_parents=False):
if self.pk is None:
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)
collector.collect([self], keep_parents=keep_parents)
return collector.delete()
description¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
full_clean¶
Call clean_fields(), clean(), and validate_unique() on the model.
Raise a ValidationError for any errors that occur.
View Source
def full_clean(self, exclude=None, validate_unique=True):
"""
Call clean_fields(), clean(), and validate_unique() on the model.
Raise a ValidationError for any errors that occur.
"""
errors = {}
if exclude is None:
exclude = []
else:
exclude = list(exclude)
try:
self.clean_fields(exclude=exclude)
except ValidationError as e:
errors = e.update_error_dict(errors)
# Form.clean() is run even if other validation fails, so do the
# same with Model.clean() for consistency.
try:
self.clean()
except ValidationError as e:
errors = e.update_error_dict(errors)
# Run unique checks, but only for fields that passed validation.
if validate_unique:
for name in errors:
if name != NON_FIELD_ERRORS and name not in exclude:
exclude.append(name)
try:
self.validate_unique(exclude=exclude)
except ValidationError as e:
errors = e.update_error_dict(errors)
if errors:
raise ValidationError(errors)
get_deferred_fields¶
Return a set containing names of deferred fields for this instance.
View Source
get_preprocessing_torch_model¶
Converts the preprocessing to a PyTorch model.
Returns:
| Type | Description |
|---|---|
| torch.nn.Module | The PyTorch model. |
View Source
get_real_concrete_instance_class¶
View Source
get_real_concrete_instance_class_id¶
View Source
get_real_instance¶
Upcast an object to it's actual type.
If a non-polymorphic manager (like base_objects) has been used to retrieve objects, then the complete object with it's real class/type and all fields may be retrieved with this method.
.. note::
Each method call executes one db query (if necessary).
Use the :meth:~polymorphic.managers.PolymorphicQuerySet.get_real_instances
to upcast a complete list in a single efficient query.
View Source
def get_real_instance(self):
"""
Upcast an object to it's actual type.
If a non-polymorphic manager (like base_objects) has been used to
retrieve objects, then the complete object with it's real class/type
and all fields may be retrieved with this method.
.. note::
Each method call executes one db query (if necessary).
Use the :meth:`~polymorphic.managers.PolymorphicQuerySet.get_real_instances`
to upcast a complete list in a single efficient query.
"""
real_model = self.get_real_instance_class()
if real_model == self.__class__:
return self
return real_model.objects.db_manager(self._state.db).get(pk=self.pk)
get_real_instance_class¶
Return the actual model type of the object.
If a non-polymorphic manager (like base_objects) has been used to retrieve objects, then the real class/type of these objects may be determined using this method.
View Source
def get_real_instance_class(self):
"""
Return the actual model type of the object.
If a non-polymorphic manager (like base_objects) has been used to
retrieve objects, then the real class/type of these objects may be
determined using this method.
"""
if self.polymorphic_ctype_id is None:
raise PolymorphicTypeUndefined(
(
"The model {}#{} does not have a `polymorphic_ctype_id` value defined.\n"
"If you created models outside polymorphic, e.g. through an import or migration, "
"make sure the `polymorphic_ctype_id` field points to the ContentType ID of the model subclass."
).format(self.__class__.__name__, self.pk)
)
# the following line would be the easiest way to do this, but it produces sql queries
# return self.polymorphic_ctype.model_class()
# so we use the following version, which uses the ContentType manager cache.
# Note that model_class() can return None for stale content types;
# when the content type record still exists but no longer refers to an existing model.
model = (
ContentType.objects.db_manager(self._state.db)
.get_for_id(self.polymorphic_ctype_id)
.model_class()
)
# Protect against bad imports (dumpdata without --natural) or other
# issues missing with the ContentType models.
if (
model is not None
and not issubclass(model, self.__class__)
and (
self.__class__._meta.proxy_for_model is None
or not issubclass(model, self.__class__._meta.proxy_for_model)
)
):
raise PolymorphicTypeInvalid(
"ContentType {} for {} #{} does not point to a subclass!".format(
self.polymorphic_ctype_id, model, self.pk
)
)
return model
get_torch_model¶
Converts the model weights to a PyTorch model.
Returns:
| Type | Description |
|---|---|
| torch.nn.Module | The PyTorch model. |
View Source
get_training¶
Gets the training associated with the model.
Returns:
| Type | Description |
|---|---|
| models.Training | The training associated with the model. |
View Source
id¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
input_shape¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
is_global_model¶
Checks if the model is a global model.
Returns:
| Type | Description |
|---|---|
| bool | True if the model is a global model, False otherwise. |
View Source
is_local_model¶
Checks if the model is a local model.
Returns:
| Type | Description |
|---|---|
| bool | True if the model is a local model, False otherwise. |
View Source
name¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
pre_save_polymorphic¶
Make sure the polymorphic_ctype value is correctly set on this model.
View Source
def pre_save_polymorphic(self, using=DEFAULT_DB_ALIAS):
"""
Make sure the ``polymorphic_ctype`` value is correctly set on this model.
"""
# This function may be called manually in special use-cases. When the object
# is saved for the first time, we store its real class in polymorphic_ctype.
# When the object later is retrieved by PolymorphicQuerySet, it uses this
# field to figure out the real class of this object
# (used by PolymorphicQuerySet._get_real_instances)
if not self.polymorphic_ctype_id:
self.polymorphic_ctype = ContentType.objects.db_manager(using).get_for_model(
self, for_concrete_model=False
)
prepare_database_save¶
View Source
preprocessing¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
refresh_from_db¶
Reload field values from the database.
By default, the reloading happens from the database this instance was loaded from, or by the read router if this instance wasn't loaded from any database. The using parameter will override the default.
Fields can be used to specify which fields to reload. The fields should be an iterable of field attnames. If fields is None, then all non-deferred fields are reloaded.
When accessing deferred fields of an instance, the deferred loading of the field will call this method.
View Source
def refresh_from_db(self, using=None, fields=None):
"""
Reload field values from the database.
By default, the reloading happens from the database this instance was
loaded from, or by the read router if this instance wasn't loaded from
any database. The using parameter will override the default.
Fields can be used to specify which fields to reload. The fields
should be an iterable of field attnames. If fields is None, then
all non-deferred fields are reloaded.
When accessing deferred fields of an instance, the deferred loading
of the field will call this method.
"""
if fields is None:
self._prefetched_objects_cache = {}
else:
prefetched_objects_cache = getattr(self, "_prefetched_objects_cache", ())
for field in fields:
if field in prefetched_objects_cache:
del prefetched_objects_cache[field]
fields.remove(field)
if not fields:
return
if any(LOOKUP_SEP in f for f in fields):
raise ValueError(
'Found "%s" in fields argument. Relations and transforms '
"are not allowed in fields." % LOOKUP_SEP
)
hints = {"instance": self}
db_instance_qs = self.__class__._base_manager.db_manager(
using, hints=hints
).filter(pk=self.pk)
# Use provided fields, if not set then reload all non-deferred fields.
deferred_fields = self.get_deferred_fields()
if fields is not None:
fields = list(fields)
db_instance_qs = db_instance_qs.only(*fields)
elif deferred_fields:
fields = [
f.attname
for f in self._meta.concrete_fields
if f.attname not in deferred_fields
]
db_instance_qs = db_instance_qs.only(*fields)
db_instance = db_instance_qs.get()
non_loaded_fields = db_instance.get_deferred_fields()
for field in self._meta.concrete_fields:
if field.attname in non_loaded_fields:
# This field wasn't refreshed - skip ahead.
continue
setattr(self, field.attname, getattr(db_instance, field.attname))
# Clear cached foreign keys.
if field.is_relation and field.is_cached(self):
field.delete_cached_value(self)
# Clear cached relations.
for field in self._meta.related_objects:
if field.is_cached(self):
field.delete_cached_value(self)
self._state.db = db_instance._state.db
round¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
save¶
Calls :meth:pre_save_polymorphic and saves the model.
View Source
save_base¶
def save_base(
self,
raw=False,
force_insert=False,
force_update=False,
using=None,
update_fields=None
)
Handle the parts of saving which should be done only once per save,
yet need to be done in raw saves, too. This includes some sanity checks and signal sending.
The 'raw' argument is telling save_base not to save any parent models and not to do any changes to the values before save. This is used by fixture loading.
View Source
def save_base(
self,
raw=False,
force_insert=False,
force_update=False,
using=None,
update_fields=None,
):
"""
Handle the parts of saving which should be done only once per save,
yet need to be done in raw saves, too. This includes some sanity
checks and signal sending.
The 'raw' argument is telling save_base not to save any parent
models and not to do any changes to the values before save. This
is used by fixture loading.
"""
using = using or router.db_for_write(self.__class__, instance=self)
assert not (force_insert and (force_update or update_fields))
assert update_fields is None or update_fields
cls = origin = self.__class__
# Skip proxies, but keep the origin as the proxy model.
if cls._meta.proxy:
cls = cls._meta.concrete_model
meta = cls._meta
if not meta.auto_created:
pre_save.send(
sender=origin,
instance=self,
raw=raw,
using=using,
update_fields=update_fields,
)
# A transaction isn't needed if one query is issued.
if meta.parents:
context_manager = transaction.atomic(using=using, savepoint=False)
else:
context_manager = transaction.mark_for_rollback_on_error(using=using)
with context_manager:
parent_inserted = False
if not raw:
parent_inserted = self._save_parents(cls, using, update_fields)
updated = self._save_table(
raw,
cls,
force_insert or parent_inserted,
force_update,
using,
update_fields,
)
# Store the database on which the object was saved
self._state.db = using
# Once saved, this is no longer a to-be-added instance.
self._state.adding = False
# Signal that the save is complete
if not meta.auto_created:
post_save.send(
sender=origin,
instance=self,
created=(not updated),
update_fields=update_fields,
raw=raw,
using=using,
)
serializable_value¶
Return the value of the field name for this instance. If the field is
a foreign key, return the id value instead of the object. If there's no Field object with this name on the model, return the model attribute's value.
Used to serialize a field's value (in the serializer, or form output, for example). Normally, you would just access the attribute directly and not use this method.
View Source
def serializable_value(self, field_name):
"""
Return the value of the field name for this instance. If the field is
a foreign key, return the id value instead of the object. If there's
no Field object with this name on the model, return the model
attribute's value.
Used to serialize a field's value (in the serializer, or form output,
for example). Normally, you would just access the attribute directly
and not use this method.
"""
try:
field = self._meta.get_field(field_name)
except FieldDoesNotExist:
return getattr(self, field_name)
return getattr(self, field.attname)
set_preprocessing_torch_model¶
Sets the preprocessing from a PyTorch model.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| value | torch.nn.Module | The PyTorch model. | None |
View Source
set_torch_model¶
Sets the model weights from a PyTorch model.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| value | torch.nn.Module | The PyTorch model. | None |
View Source
swag_first_moment¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
swag_second_moment¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
unique_error_message¶
View Source
def unique_error_message(self, model_class, unique_check):
opts = model_class._meta
params = {
"model": self,
"model_class": model_class,
"model_name": capfirst(opts.verbose_name),
"unique_check": unique_check,
}
# A unique field
if len(unique_check) == 1:
field = opts.get_field(unique_check[0])
params["field_label"] = capfirst(field.verbose_name)
return ValidationError(
message=field.error_messages["unique"],
code="unique",
params=params,
)
# unique_together
else:
field_labels = [
capfirst(opts.get_field(f).verbose_name) for f in unique_check
]
params["field_labels"] = get_text_list(field_labels, _("and"))
return ValidationError(
message=_("%(model_name)s with this %(field_labels)s already exists."),
code="unique_together",
params=params,
)
validate_unique¶
Check unique constraints on the model and raise ValidationError if any
failed.
View Source
def validate_unique(self, exclude=None):
"""
Check unique constraints on the model and raise ValidationError if any
failed.
"""
unique_checks, date_checks = self._get_unique_checks(exclude=exclude)
errors = self._perform_unique_checks(unique_checks)
date_errors = self._perform_date_checks(date_checks)
for k, v in date_errors.items():
errors.setdefault(k, []).extend(v)
if errors:
raise ValidationError(errors)
weights¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
Training¶
Training model class.
View Source
class Training(models.Model):
"""
Training model class.
"""
id: UUIDField = UUIDField(primary_key=True, editable=False, default=uuid4)
"""Unique identifier for the training."""
model: OneToOneField = OneToOneField(Model, on_delete=CASCADE)
"""Model used in the training."""
actor: ForeignKey = ForeignKey(User, on_delete=CASCADE, related_name="actors")
"""User who is the actor of the training."""
participants: ManyToManyField = ManyToManyField(User)
"""Users who are the participants of the training."""
state: CharField = CharField(max_length=1, choices=TrainingState.choices)
"""State of the training."""
target_num_updates: IntegerField = IntegerField()
"""Target number of updates for the training."""
last_update = models.DateTimeField(auto_now=True)
"""Time of the last update."""
uncertainty_method: CharField = CharField(
max_length=32, choices=UncertaintyMethod.choices, default=UncertaintyMethod.NONE
)
"""Uncertainty method used in the training."""
aggregation_method: CharField = CharField(
max_length=32, choices=AggregationMethod.choices, default=AggregationMethod.FED_AVG
)
"""Aggregation method used in the training."""
# HINT: https://docs.djangoproject.com/en/4.2/topics/db/queries/#querying-jsonfield
options: JSONField = JSONField(default=dict, encoder=DjangoJSONEncoder)
"""Options for the training."""
locked: BooleanField = BooleanField(default=False)
"""Flag indicating whether the training is locked."""
Ancestors (in MRO)¶
- django.db.models.base.Model
Class variables¶
Time of the last update.
Static methods¶
check¶
View Source
@classmethod
def check(cls, **kwargs):
errors = [
*cls._check_swappable(),
*cls._check_model(),
*cls._check_managers(**kwargs),
]
if not cls._meta.swapped:
databases = kwargs.get("databases") or []
errors += [
*cls._check_fields(**kwargs),
*cls._check_m2m_through_same_relationship(),
*cls._check_long_column_names(databases),
]
clash_errors = (
*cls._check_id_field(),
*cls._check_field_name_clashes(),
*cls._check_model_name_db_lookup_clashes(),
*cls._check_property_name_related_field_accessor_clashes(),
*cls._check_single_primary_key(),
)
errors.extend(clash_errors)
# If there are field name clashes, hide consequent column name
# clashes.
if not clash_errors:
errors.extend(cls._check_column_name_clashes())
errors += [
*cls._check_index_together(),
*cls._check_unique_together(),
*cls._check_indexes(databases),
*cls._check_ordering(),
*cls._check_constraints(databases),
*cls._check_default_pk(),
]
return errors
from_db¶
View Source
@classmethod
def from_db(cls, db, field_names, values):
if len(values) != len(cls._meta.concrete_fields):
values_iter = iter(values)
values = [
next(values_iter) if f.attname in field_names else DEFERRED
for f in cls._meta.concrete_fields
]
new = cls(*values)
new._state.adding = False
new._state.db = db
return new
Instance variables¶
Methods¶
aggregation_method¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
clean¶
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.
View Source
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
clean_fields¶
Clean all fields and raise a ValidationError containing a dict
of all validation errors if any occur.
View Source
def clean_fields(self, exclude=None):
"""
Clean all fields and raise a ValidationError containing a dict
of all validation errors if any occur.
"""
if exclude is None:
exclude = []
errors = {}
for f in self._meta.fields:
if f.name in exclude:
continue
# Skip validation for empty fields with blank=True. The developer
# is responsible for making sure they have a valid value.
raw_value = getattr(self, f.attname)
if f.blank and raw_value in f.empty_values:
continue
try:
setattr(self, f.attname, f.clean(raw_value, self))
except ValidationError as e:
errors[f.name] = e.error_list
if errors:
raise ValidationError(errors)
date_error_message¶
View Source
def date_error_message(self, lookup_type, field_name, unique_for):
opts = self._meta
field = opts.get_field(field_name)
return ValidationError(
message=field.error_messages["unique_for_date"],
code="unique_for_date",
params={
"model": self,
"model_name": capfirst(opts.verbose_name),
"lookup_type": lookup_type,
"field": field_name,
"field_label": capfirst(field.verbose_name),
"date_field": unique_for,
"date_field_label": capfirst(opts.get_field(unique_for).verbose_name),
},
)
delete¶
View Source
def delete(self, using=None, keep_parents=False):
if self.pk is None:
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)
collector.collect([self], keep_parents=keep_parents)
return collector.delete()
full_clean¶
Call clean_fields(), clean(), and validate_unique() on the model.
Raise a ValidationError for any errors that occur.
View Source
def full_clean(self, exclude=None, validate_unique=True):
"""
Call clean_fields(), clean(), and validate_unique() on the model.
Raise a ValidationError for any errors that occur.
"""
errors = {}
if exclude is None:
exclude = []
else:
exclude = list(exclude)
try:
self.clean_fields(exclude=exclude)
except ValidationError as e:
errors = e.update_error_dict(errors)
# Form.clean() is run even if other validation fails, so do the
# same with Model.clean() for consistency.
try:
self.clean()
except ValidationError as e:
errors = e.update_error_dict(errors)
# Run unique checks, but only for fields that passed validation.
if validate_unique:
for name in errors:
if name != NON_FIELD_ERRORS and name not in exclude:
exclude.append(name)
try:
self.validate_unique(exclude=exclude)
except ValidationError as e:
errors = e.update_error_dict(errors)
if errors:
raise ValidationError(errors)
get_aggregation_method_display¶
def get_aggregation_method_display(
self,
*,
field=<django.db.models.fields.CharField: aggregation_method>
)
View Source
get_deferred_fields¶
Return a set containing names of deferred fields for this instance.
View Source
get_next_by_last_update¶
def get_next_by_last_update(
self,
*,
field=<django.db.models.fields.DateTimeField: last_update>,
is_next=True,
**kwargs
)
View Source
get_previous_by_last_update¶
def get_previous_by_last_update(
self,
*,
field=<django.db.models.fields.DateTimeField: last_update>,
is_next=False,
**kwargs
)
View Source
get_state_display¶
View Source
get_uncertainty_method_display¶
def get_uncertainty_method_display(
self,
*,
field=<django.db.models.fields.CharField: uncertainty_method>
)
View Source
id¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
locked¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
options¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
prepare_database_save¶
View Source
refresh_from_db¶
Reload field values from the database.
By default, the reloading happens from the database this instance was loaded from, or by the read router if this instance wasn't loaded from any database. The using parameter will override the default.
Fields can be used to specify which fields to reload. The fields should be an iterable of field attnames. If fields is None, then all non-deferred fields are reloaded.
When accessing deferred fields of an instance, the deferred loading of the field will call this method.
View Source
def refresh_from_db(self, using=None, fields=None):
"""
Reload field values from the database.
By default, the reloading happens from the database this instance was
loaded from, or by the read router if this instance wasn't loaded from
any database. The using parameter will override the default.
Fields can be used to specify which fields to reload. The fields
should be an iterable of field attnames. If fields is None, then
all non-deferred fields are reloaded.
When accessing deferred fields of an instance, the deferred loading
of the field will call this method.
"""
if fields is None:
self._prefetched_objects_cache = {}
else:
prefetched_objects_cache = getattr(self, "_prefetched_objects_cache", ())
for field in fields:
if field in prefetched_objects_cache:
del prefetched_objects_cache[field]
fields.remove(field)
if not fields:
return
if any(LOOKUP_SEP in f for f in fields):
raise ValueError(
'Found "%s" in fields argument. Relations and transforms '
"are not allowed in fields." % LOOKUP_SEP
)
hints = {"instance": self}
db_instance_qs = self.__class__._base_manager.db_manager(
using, hints=hints
).filter(pk=self.pk)
# Use provided fields, if not set then reload all non-deferred fields.
deferred_fields = self.get_deferred_fields()
if fields is not None:
fields = list(fields)
db_instance_qs = db_instance_qs.only(*fields)
elif deferred_fields:
fields = [
f.attname
for f in self._meta.concrete_fields
if f.attname not in deferred_fields
]
db_instance_qs = db_instance_qs.only(*fields)
db_instance = db_instance_qs.get()
non_loaded_fields = db_instance.get_deferred_fields()
for field in self._meta.concrete_fields:
if field.attname in non_loaded_fields:
# This field wasn't refreshed - skip ahead.
continue
setattr(self, field.attname, getattr(db_instance, field.attname))
# Clear cached foreign keys.
if field.is_relation and field.is_cached(self):
field.delete_cached_value(self)
# Clear cached relations.
for field in self._meta.related_objects:
if field.is_cached(self):
field.delete_cached_value(self)
self._state.db = db_instance._state.db
save¶
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.
View Source
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_fields = self.get_deferred_fields()
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 = set()
for field in self._meta.concrete_fields:
if not field.primary_key:
field_names.add(field.name)
if field.name != field.attname:
field_names.add(field.attname)
non_model_fields = update_fields.difference(field_names)
if non_model_fields:
raise ValueError(
"The following fields do not exist in this model, are m2m "
"fields, or are non-concrete fields: %s"
% ", ".join(non_model_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_fields and using == self._state.db:
field_names = set()
for field in self._meta.concrete_fields:
if not field.primary_key and not hasattr(field, "through"):
field_names.add(field.attname)
loaded_fields = field_names.difference(deferred_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,
)
save_base¶
def save_base(
self,
raw=False,
force_insert=False,
force_update=False,
using=None,
update_fields=None
)
Handle the parts of saving which should be done only once per save,
yet need to be done in raw saves, too. This includes some sanity checks and signal sending.
The 'raw' argument is telling save_base not to save any parent models and not to do any changes to the values before save. This is used by fixture loading.
View Source
def save_base(
self,
raw=False,
force_insert=False,
force_update=False,
using=None,
update_fields=None,
):
"""
Handle the parts of saving which should be done only once per save,
yet need to be done in raw saves, too. This includes some sanity
checks and signal sending.
The 'raw' argument is telling save_base not to save any parent
models and not to do any changes to the values before save. This
is used by fixture loading.
"""
using = using or router.db_for_write(self.__class__, instance=self)
assert not (force_insert and (force_update or update_fields))
assert update_fields is None or update_fields
cls = origin = self.__class__
# Skip proxies, but keep the origin as the proxy model.
if cls._meta.proxy:
cls = cls._meta.concrete_model
meta = cls._meta
if not meta.auto_created:
pre_save.send(
sender=origin,
instance=self,
raw=raw,
using=using,
update_fields=update_fields,
)
# A transaction isn't needed if one query is issued.
if meta.parents:
context_manager = transaction.atomic(using=using, savepoint=False)
else:
context_manager = transaction.mark_for_rollback_on_error(using=using)
with context_manager:
parent_inserted = False
if not raw:
parent_inserted = self._save_parents(cls, using, update_fields)
updated = self._save_table(
raw,
cls,
force_insert or parent_inserted,
force_update,
using,
update_fields,
)
# Store the database on which the object was saved
self._state.db = using
# Once saved, this is no longer a to-be-added instance.
self._state.adding = False
# Signal that the save is complete
if not meta.auto_created:
post_save.send(
sender=origin,
instance=self,
created=(not updated),
update_fields=update_fields,
raw=raw,
using=using,
)
serializable_value¶
Return the value of the field name for this instance. If the field is
a foreign key, return the id value instead of the object. If there's no Field object with this name on the model, return the model attribute's value.
Used to serialize a field's value (in the serializer, or form output, for example). Normally, you would just access the attribute directly and not use this method.
View Source
def serializable_value(self, field_name):
"""
Return the value of the field name for this instance. If the field is
a foreign key, return the id value instead of the object. If there's
no Field object with this name on the model, return the model
attribute's value.
Used to serialize a field's value (in the serializer, or form output,
for example). Normally, you would just access the attribute directly
and not use this method.
"""
try:
field = self._meta.get_field(field_name)
except FieldDoesNotExist:
return getattr(self, field_name)
return getattr(self, field.attname)
state¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
target_num_updates¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
uncertainty_method¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
unique_error_message¶
View Source
def unique_error_message(self, model_class, unique_check):
opts = model_class._meta
params = {
"model": self,
"model_class": model_class,
"model_name": capfirst(opts.verbose_name),
"unique_check": unique_check,
}
# A unique field
if len(unique_check) == 1:
field = opts.get_field(unique_check[0])
params["field_label"] = capfirst(field.verbose_name)
return ValidationError(
message=field.error_messages["unique"],
code="unique",
params=params,
)
# unique_together
else:
field_labels = [
capfirst(opts.get_field(f).verbose_name) for f in unique_check
]
params["field_labels"] = get_text_list(field_labels, _("and"))
return ValidationError(
message=_("%(model_name)s with this %(field_labels)s already exists."),
code="unique_together",
params=params,
)
validate_unique¶
Check unique constraints on the model and raise ValidationError if any
failed.
View Source
def validate_unique(self, exclude=None):
"""
Check unique constraints on the model and raise ValidationError if any
failed.
"""
unique_checks, date_checks = self._get_unique_checks(exclude=exclude)
errors = self._perform_unique_checks(unique_checks)
date_errors = self._perform_date_checks(date_checks)
for k, v in date_errors.items():
errors.setdefault(k, []).extend(v)
if errors:
raise ValidationError(errors)
User¶
User class.
Inherits from Django's AbstractUser and NotificationReceiver.
View Source
class User(AbstractUser, NotificationReceiver):
"""
User class.
Inherits from Django's AbstractUser and NotificationReceiver.
"""
id: UUIDField = UUIDField(primary_key=True, editable=False, default=uuid4)
"""Unique identifier for the user."""
actor: BooleanField = BooleanField(default=False)
"""Flag indicating whether the user is an actor."""
client: BooleanField = BooleanField(default=False)
"""Flag indicating whether the user is a client."""
message_endpoint: URLField = URLField()
"""Endpoint to send the message to."""
Ancestors (in MRO)¶
- django.contrib.auth.models.AbstractUser
- django.contrib.auth.base_user.AbstractBaseUser
- django.contrib.auth.models.PermissionsMixin
- django.db.models.base.Model
- fl_server_core.models.user.NotificationReceiver
Class variables¶
Static methods¶
check¶
View Source
@classmethod
def check(cls, **kwargs):
errors = [
*cls._check_swappable(),
*cls._check_model(),
*cls._check_managers(**kwargs),
]
if not cls._meta.swapped:
databases = kwargs.get("databases") or []
errors += [
*cls._check_fields(**kwargs),
*cls._check_m2m_through_same_relationship(),
*cls._check_long_column_names(databases),
]
clash_errors = (
*cls._check_id_field(),
*cls._check_field_name_clashes(),
*cls._check_model_name_db_lookup_clashes(),
*cls._check_property_name_related_field_accessor_clashes(),
*cls._check_single_primary_key(),
)
errors.extend(clash_errors)
# If there are field name clashes, hide consequent column name
# clashes.
if not clash_errors:
errors.extend(cls._check_column_name_clashes())
errors += [
*cls._check_index_together(),
*cls._check_unique_together(),
*cls._check_indexes(databases),
*cls._check_ordering(),
*cls._check_constraints(databases),
*cls._check_default_pk(),
]
return errors
from_db¶
View Source
@classmethod
def from_db(cls, db, field_names, values):
if len(values) != len(cls._meta.concrete_fields):
values_iter = iter(values)
values = [
next(values_iter) if f.attname in field_names else DEFERRED
for f in cls._meta.concrete_fields
]
new = cls(*values)
new._state.adding = False
new._state.db = db
return new
get_email_field_name¶
View Source
normalize_username¶
View Source
Instance variables¶
Always return False. This is a way of comparing User objects to
anonymous users.
Always return True. This is a way to tell if the user has been
authenticated in templates.
Methods¶
actor¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
check_password¶
Return a boolean of whether the raw_password was correct. Handles
hashing formats behind the scenes.
View Source
def check_password(self, raw_password):
"""
Return a boolean of whether the raw_password was correct. Handles
hashing formats behind the scenes.
"""
def setter(raw_password):
self.set_password(raw_password)
# Password hash upgrades shouldn't be considered password changes.
self._password = None
self.save(update_fields=["password"])
return check_password(raw_password, self.password, setter)
clean¶
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.
View Source
clean_fields¶
Clean all fields and raise a ValidationError containing a dict
of all validation errors if any occur.
View Source
def clean_fields(self, exclude=None):
"""
Clean all fields and raise a ValidationError containing a dict
of all validation errors if any occur.
"""
if exclude is None:
exclude = []
errors = {}
for f in self._meta.fields:
if f.name in exclude:
continue
# Skip validation for empty fields with blank=True. The developer
# is responsible for making sure they have a valid value.
raw_value = getattr(self, f.attname)
if f.blank and raw_value in f.empty_values:
continue
try:
setattr(self, f.attname, f.clean(raw_value, self))
except ValidationError as e:
errors[f.name] = e.error_list
if errors:
raise ValidationError(errors)
client¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
date_error_message¶
View Source
def date_error_message(self, lookup_type, field_name, unique_for):
opts = self._meta
field = opts.get_field(field_name)
return ValidationError(
message=field.error_messages["unique_for_date"],
code="unique_for_date",
params={
"model": self,
"model_name": capfirst(opts.verbose_name),
"lookup_type": lookup_type,
"field": field_name,
"field_label": capfirst(field.verbose_name),
"date_field": unique_for,
"date_field_label": capfirst(opts.get_field(unique_for).verbose_name),
},
)
date_joined¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
delete¶
View Source
def delete(self, using=None, keep_parents=False):
if self.pk is None:
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)
collector.collect([self], keep_parents=keep_parents)
return collector.delete()
email¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
email_user¶
Send an email to this user.
View Source
first_name¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
full_clean¶
Call clean_fields(), clean(), and validate_unique() on the model.
Raise a ValidationError for any errors that occur.
View Source
def full_clean(self, exclude=None, validate_unique=True):
"""
Call clean_fields(), clean(), and validate_unique() on the model.
Raise a ValidationError for any errors that occur.
"""
errors = {}
if exclude is None:
exclude = []
else:
exclude = list(exclude)
try:
self.clean_fields(exclude=exclude)
except ValidationError as e:
errors = e.update_error_dict(errors)
# Form.clean() is run even if other validation fails, so do the
# same with Model.clean() for consistency.
try:
self.clean()
except ValidationError as e:
errors = e.update_error_dict(errors)
# Run unique checks, but only for fields that passed validation.
if validate_unique:
for name in errors:
if name != NON_FIELD_ERRORS and name not in exclude:
exclude.append(name)
try:
self.validate_unique(exclude=exclude)
except ValidationError as e:
errors = e.update_error_dict(errors)
if errors:
raise ValidationError(errors)
get_all_permissions¶
get_deferred_fields¶
Return a set containing names of deferred fields for this instance.
View Source
get_full_name¶
Return the first_name plus the last_name, with a space in between.
View Source
get_group_permissions¶
Return a list of permission strings that this user has through their
groups. Query all available auth backends. If an object is passed in, return only permissions matching this object.
View Source
get_next_by_date_joined¶
def get_next_by_date_joined(
self,
*,
field=<django.db.models.fields.DateTimeField: date_joined>,
is_next=True,
**kwargs
)
View Source
get_previous_by_date_joined¶
def get_previous_by_date_joined(
self,
*,
field=<django.db.models.fields.DateTimeField: date_joined>,
is_next=False,
**kwargs
)
View Source
get_session_auth_hash¶
Return an HMAC of the password field.
View Source
get_short_name¶
Return the short name for the user.
View Source
get_user_permissions¶
Return a list of permission strings that this user has directly.
Query all available auth backends. If an object is passed in, return only permissions matching this object.
View Source
get_username¶
Return the username for this User.
View Source
has_module_perms¶
Return True if the user has any permissions in the given app label.
Use similar logic as has_perm(), above.
View Source
has_perm¶
Return True if the user has the specified permission. Query all
available auth backends, but return immediately if any backend returns True. Thus, a user who has permission from a single auth backend is assumed to have permission in general. If an object is provided, check permissions for that object.
View Source
def has_perm(self, perm, obj=None):
"""
Return True if the user has the specified permission. Query all
available auth backends, but return immediately if any backend returns
True. Thus, a user who has permission from a single auth backend is
assumed to have permission in general. If an object is provided, check
permissions for that object.
"""
# Active superusers have all permissions.
if self.is_active and self.is_superuser:
return True
# Otherwise we need to check the backends.
return _user_has_perm(self, perm, obj)
has_perms¶
Return True if the user has each of the specified permissions. If
object is passed, check if the user has all required perms for it.
View Source
has_usable_password¶
Return False if set_unusable_password() has been called for this user.
View Source
id¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
is_active¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
is_staff¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
is_superuser¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
last_login¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
last_name¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
message_endpoint¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
natural_key¶
password¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
prepare_database_save¶
View Source
refresh_from_db¶
Reload field values from the database.
By default, the reloading happens from the database this instance was loaded from, or by the read router if this instance wasn't loaded from any database. The using parameter will override the default.
Fields can be used to specify which fields to reload. The fields should be an iterable of field attnames. If fields is None, then all non-deferred fields are reloaded.
When accessing deferred fields of an instance, the deferred loading of the field will call this method.
View Source
def refresh_from_db(self, using=None, fields=None):
"""
Reload field values from the database.
By default, the reloading happens from the database this instance was
loaded from, or by the read router if this instance wasn't loaded from
any database. The using parameter will override the default.
Fields can be used to specify which fields to reload. The fields
should be an iterable of field attnames. If fields is None, then
all non-deferred fields are reloaded.
When accessing deferred fields of an instance, the deferred loading
of the field will call this method.
"""
if fields is None:
self._prefetched_objects_cache = {}
else:
prefetched_objects_cache = getattr(self, "_prefetched_objects_cache", ())
for field in fields:
if field in prefetched_objects_cache:
del prefetched_objects_cache[field]
fields.remove(field)
if not fields:
return
if any(LOOKUP_SEP in f for f in fields):
raise ValueError(
'Found "%s" in fields argument. Relations and transforms '
"are not allowed in fields." % LOOKUP_SEP
)
hints = {"instance": self}
db_instance_qs = self.__class__._base_manager.db_manager(
using, hints=hints
).filter(pk=self.pk)
# Use provided fields, if not set then reload all non-deferred fields.
deferred_fields = self.get_deferred_fields()
if fields is not None:
fields = list(fields)
db_instance_qs = db_instance_qs.only(*fields)
elif deferred_fields:
fields = [
f.attname
for f in self._meta.concrete_fields
if f.attname not in deferred_fields
]
db_instance_qs = db_instance_qs.only(*fields)
db_instance = db_instance_qs.get()
non_loaded_fields = db_instance.get_deferred_fields()
for field in self._meta.concrete_fields:
if field.attname in non_loaded_fields:
# This field wasn't refreshed - skip ahead.
continue
setattr(self, field.attname, getattr(db_instance, field.attname))
# Clear cached foreign keys.
if field.is_relation and field.is_cached(self):
field.delete_cached_value(self)
# Clear cached relations.
for field in self._meta.related_objects:
if field.is_cached(self):
field.delete_cached_value(self)
self._state.db = db_instance._state.db
save¶
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.
View Source
save_base¶
def save_base(
self,
raw=False,
force_insert=False,
force_update=False,
using=None,
update_fields=None
)
Handle the parts of saving which should be done only once per save,
yet need to be done in raw saves, too. This includes some sanity checks and signal sending.
The 'raw' argument is telling save_base not to save any parent models and not to do any changes to the values before save. This is used by fixture loading.
View Source
def save_base(
self,
raw=False,
force_insert=False,
force_update=False,
using=None,
update_fields=None,
):
"""
Handle the parts of saving which should be done only once per save,
yet need to be done in raw saves, too. This includes some sanity
checks and signal sending.
The 'raw' argument is telling save_base not to save any parent
models and not to do any changes to the values before save. This
is used by fixture loading.
"""
using = using or router.db_for_write(self.__class__, instance=self)
assert not (force_insert and (force_update or update_fields))
assert update_fields is None or update_fields
cls = origin = self.__class__
# Skip proxies, but keep the origin as the proxy model.
if cls._meta.proxy:
cls = cls._meta.concrete_model
meta = cls._meta
if not meta.auto_created:
pre_save.send(
sender=origin,
instance=self,
raw=raw,
using=using,
update_fields=update_fields,
)
# A transaction isn't needed if one query is issued.
if meta.parents:
context_manager = transaction.atomic(using=using, savepoint=False)
else:
context_manager = transaction.mark_for_rollback_on_error(using=using)
with context_manager:
parent_inserted = False
if not raw:
parent_inserted = self._save_parents(cls, using, update_fields)
updated = self._save_table(
raw,
cls,
force_insert or parent_inserted,
force_update,
using,
update_fields,
)
# Store the database on which the object was saved
self._state.db = using
# Once saved, this is no longer a to-be-added instance.
self._state.adding = False
# Signal that the save is complete
if not meta.auto_created:
post_save.send(
sender=origin,
instance=self,
created=(not updated),
update_fields=update_fields,
raw=raw,
using=using,
)
serializable_value¶
Return the value of the field name for this instance. If the field is
a foreign key, return the id value instead of the object. If there's no Field object with this name on the model, return the model attribute's value.
Used to serialize a field's value (in the serializer, or form output, for example). Normally, you would just access the attribute directly and not use this method.
View Source
def serializable_value(self, field_name):
"""
Return the value of the field name for this instance. If the field is
a foreign key, return the id value instead of the object. If there's
no Field object with this name on the model, return the model
attribute's value.
Used to serialize a field's value (in the serializer, or form output,
for example). Normally, you would just access the attribute directly
and not use this method.
"""
try:
field = self._meta.get_field(field_name)
except FieldDoesNotExist:
return getattr(self, field_name)
return getattr(self, field.attname)
set_password¶
View Source
set_unusable_password¶
View Source
unique_error_message¶
View Source
def unique_error_message(self, model_class, unique_check):
opts = model_class._meta
params = {
"model": self,
"model_class": model_class,
"model_name": capfirst(opts.verbose_name),
"unique_check": unique_check,
}
# A unique field
if len(unique_check) == 1:
field = opts.get_field(unique_check[0])
params["field_label"] = capfirst(field.verbose_name)
return ValidationError(
message=field.error_messages["unique"],
code="unique",
params=params,
)
# unique_together
else:
field_labels = [
capfirst(opts.get_field(f).verbose_name) for f in unique_check
]
params["field_labels"] = get_text_list(field_labels, _("and"))
return ValidationError(
message=_("%(model_name)s with this %(field_labels)s already exists."),
code="unique_together",
params=params,
)
username¶
A wrapper for a deferred-loading field. When the value is read from this
object the first time, the query is executed.
validate_unique¶
Check unique constraints on the model and raise ValidationError if any
failed.
View Source
def validate_unique(self, exclude=None):
"""
Check unique constraints on the model and raise ValidationError if any
failed.
"""
unique_checks, date_checks = self._get_unique_checks(exclude=exclude)
errors = self._perform_unique_checks(unique_checks)
date_errors = self._perform_date_checks(date_checks)
for k, v in date_errors.items():
errors.setdefault(k, []).extend(v)
if errors:
raise ValidationError(errors)