class CheckMixin(
BoolGroupMixin,
ChoiceGroupMixin,
DateGroupMixin,
FileGroupMixin,
IDGroupMixin,
ImgGroupMixin,
NumberGroupMixin,
PasswordGroupMixin,
SlugGroupMixin,
TextGroupMixin,
):
"""Validation of Model data before saving to the database."""
async def check(
self,
is_save: bool = False,
collection: AsyncCollection | None = None,
is_migration_process: bool = False,
) -> dict[str, Any]:
"""Validation of Model data before saving to the database.
It is also used to verify Models that do not migrate to the database.
"""
metadata = self.__class__.META
# Get the document ID.
doc_id: ObjectId | None = self.id
# Does the document exist in the database?
is_update: bool = doc_id is not None
# Create an identifier for a new document.
if is_save and not is_update:
doc_id = ObjectId()
setattr(self, "id", doc_id) # ruff:ignore[set-attr-with-constant]
result_map: dict[str, Any] = {}
# Errors from additional validation of fields.
error_map: NamedTuple = await self.add_validation()
# Get Model collection.
if collection is None:
collection = Config.MONGO_DATABASE[metadata["collection_name"]]
# Create params for *_group methods.
params: dict[str, Any] = {
"doc_id": doc_id,
"is_save": is_save,
"is_update": is_update, # Does the document exist in the database?
"is_error_symptom": False, # Is there any incorrect data?
"result_map": result_map, # Data to save or update to the database.
"collection": collection,
"field_core": None,
"full_model_name": metadata["full_model_name"],
"is_migration_process": is_migration_process,
"curr_doc": (await collection.find_one({"_id": doc_id}) if is_save and is_update else None),
"descriptor_fields": metadata["all_descriptor_fields"],
"LANGUAGES": self._LANGUAGES,
"LANG_CODE": self._LANG_CODE,
"_": self._RAMIFICE_TRANSLATOR.gettext,
}
# Run checking fields.
for f_name in params["descriptor_fields"]:
f__core = getattr(self, f"{f_name}__core")
# Reset a field errors to exclude duplicates.
f__core.errors = []
# Check additional validation.
err_msg = error_map[f_name]
if err_msg is not None:
f__core.errors.append(err_msg)
if not params["is_error_symptom"]:
params["is_error_symptom"] = True
# Checking the fields by groups.
if not f__core.is_ignore:
params["field_core"] = f__core
match f__core.group:
case "text":
await self.text_group(params)
case "number":
await self.number_group(params)
case "date":
self.date_group(params)
case "img":
await self.img_group(params)
case "file":
await self.file_group(params)
case "choice":
self.choice_group(params)
case "bool":
self.bool_group(params)
case "id":
self.id_group(params)
case "slug":
await self.slug_group(params)
case "password":
self.password_group(params)
case _ as unreachable:
err_msg: str = f"Unacceptable group `{unreachable}`!"
logger.critical(err_msg)
assert_never(unreachable)
# Actions in case of error.
if is_save:
if params["is_error_symptom"]:
# Reset the ObjectId for a new document.
if not is_update:
# pyrefly: ignore [bad-assignment]
self.__dict__["id"] = None
# Delete orphaned files.
curr_doc: dict[str, Any] | None = params["curr_doc"]
for field_name in params["descriptor_fields"]:
f__core = getattr(self, f"{field_name}__core")
match f__core.group:
case "file":
file_data = result_map.get(field_name)
if file_data is not None:
if file_data["is_new_file"]:
await to_thread.run_sync(remove, file_data["path"])
f__core.value = None
setattr(self, field_name, None)
if curr_doc is not None:
f__core.value = curr_doc[field_name]
setattr(self, field_name, curr_doc[field_name])
case "img":
img_data = result_map.get(field_name)
if img_data is not None:
if img_data["is_new_img"]:
# pyrefly: ignore [incompatible-overload-residual]
await to_thread.run_sync(rmtree, img_data["imgs_dir_path"])
f__core.value = None
setattr(self, field_name, None)
if curr_doc is not None:
f__core.value = curr_doc[field_name]
setattr(self, field_name, curr_doc[field_name])
else:
for field_name in params["descriptor_fields"]:
f__core = getattr(self, f"{field_name}__core")
if f__core.is_ignore:
continue
match f__core.group:
case "file":
file_data = result_map.get(field_name)
if file_data is not None:
file_data["is_new_file"] = False
case "img":
img_data = result_map.get(field_name)
if img_data is not None:
img_data["is_new_img"] = False
#
return {
"data": result_map,
"is_valid": not params["is_error_symptom"],
"is_update": is_update,
}