from django.db import models from django.utils.timezone import now from django.conf import settings from typing import List, Set from django.apps import apps class BaseModel(models.Model): creation_date = models.DateTimeField(default=now, editable=False) last_update = models.DateTimeField(default=now) related_user = models.ForeignKey(settings.AUTH_USER_MODEL, blank=True, null=True, on_delete=models.SET_NULL, related_name='+') last_updated_by = models.ForeignKey(settings.AUTH_USER_MODEL, blank=True, null=True, on_delete=models.SET_NULL, related_name='+') data_access_ids = models.JSONField(default=list) class Meta: abstract = True def save(self, *args, **kwargs): if self.related_user is None: self.related_user = self.find_related_user() if self._state.adding: self.update_data_access_list() super().save(*args, **kwargs) def get_store_id(self): if isinstance(self, SideStoreModel): return self.store_id else: return None def update_data_access_list(self): related_instances = self.related_instances() related_ids = [ri.id for ri in related_instances] related_ids.append(self.id) DataAccess = apps.get_model('sync', 'DataAccess') data_accesses = DataAccess.objects.filter(model_id__in=related_ids) for data_access in data_accesses: self.add_data_access_relation(data_access) # add data_access to children who might not had the relationship # if data_accesses: # for child in self.get_children_by_model(): # if len(child.data_access_ids) == 0: # for data_access in data_accesses: # self.add_data_access_relation(data_access) def add_data_access_relation(self, data_access): str_id = str(data_access.id) if str_id not in self.data_access_ids: self.data_access_ids.append(str_id) def remove_data_access_relation(self, data_access): self.data_access_ids.remove(str(data_access.id)) def get_children_by_model(self): """ Returns a dictionary where: - keys are the model names - values are QuerySets of related objects """ children = self._meta.get_fields(include_parents=False, include_hidden=False) related_objects = {} for child in children: if (child.one_to_many or child.one_to_one) and child.auto_created: model_name = child.related_model.__name__ related_objects[model_name] = getattr(self, child.name).all() return related_objects def get_parents_by_model(self): """ Returns a dictionary where: - keys are the model names - values are the parent model instances """ parents = {} # Get all fields including parent links fields = self._meta.get_fields(include_parents=True, include_hidden=True) for field in fields: # Check if the field is a parent link (OneToOne field that points to a parent model) if isinstance(field, models.OneToOneRel) and field.parent_link: model_name = field.related_model.__name__ # Get the parent instance using the related name parent_instance = getattr(self, field.get_accessor_name()) if parent_instance: # print(f'>>> add parent for OneToOneRel : {model_name}') parents[model_name] = parent_instance # Also check for direct foreign key relationships that might represent parent relationships elif isinstance(field, models.ForeignKey): model_name = field.related_model.__name__ parent_instance = getattr(self, field.name) if parent_instance: # print(f'>>> add parent for ForeignKey : {model_name}') parents[model_name] = parent_instance return parents def get_recursive_children(self, processed_objects: Set = None) -> List: """ Recursively get all children objects through the hierarchy """ if processed_objects is None: processed_objects = set() # Skip if we've already processed this object to avoid infinite recursion if self.pk in processed_objects: return [] processed_objects.add(self.pk) children = [] # Get immediate children children_by_model = self.get_children_by_model() for queryset in children_by_model.values(): for child in queryset: children.append(child) # Recursively get children of children if isinstance(child, BaseModel): children.extend(child.get_recursive_children(processed_objects)) return children def get_recursive_parents(self, processed_objects: Set = None) -> List: """ Recursively get all parent objects through the hierarchy """ if processed_objects is None: processed_objects = set() # Skip if we've already processed this object to avoid infinite recursion if self.pk in processed_objects: return [] processed_objects.add(self.pk) parents = [] # Get immediate parents parents_by_model = self.get_parents_by_model() for parent in parents_by_model.values(): parents.append(parent) # Recursively get parents of parents if isinstance(parent, BaseModel): parents.extend(parent.get_recursive_parents(processed_objects)) return parents def related_instances(self): """ Get all related instances (both children and parents) recursively """ instances = [] processed_objects = set() instances.extend(self.get_recursive_children(processed_objects)) processed_objects = set() instances.extend(self.get_recursive_parents(processed_objects)) return instances def find_related_user(self, processed_objects: Set = None): if processed_objects is None: processed_objects = set() # Skip if we've already processed this object to avoid infinite recursion if self.pk in processed_objects: return None processed_objects.add(self.pk) # Get immediate parents parents_by_model = self.get_parents_by_model() for parent in parents_by_model.values(): if isinstance(parent, BaseModel): if parent.related_user: print(f'related_user found in {parent}') return parent.related_user else: return parent.find_related_user(processed_objects) return None class SideStoreModel(BaseModel): store_id = models.CharField(max_length=100, default="") # a value matching LeStorage directory sub-stores. Matches the name of the directory. class Meta: abstract = True