staxriver
/
padelclub_backend
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add mecasnism to get shared instance reverse paths

Browse Source
sync3
Laurent 5 months ago
parent 90e7f4216e
commit e8d92d1216
5 changed files with 216 additions and 191 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 194
      sync/model_manager.py
  2. 23
      sync/models/base.py
  3. 3
      sync/models/data_access.py
  4. 185
      sync/registry.py
  5. 2
      tournaments/models/event.py

194
sync/model_manager.py
Unescape View File

@ -0,0 +1,194 @@
from django.conf import settings
from typing import List, Dict
from .registry import model_registry
import logging
logger = logging.getLogger(__name__)
class SyncModelChildrenManager:
"""
Manager class for handling model children sharing configuration.
Reads the SYNC_MODEL_CHILDREN_SHARING setting once and builds a bidirectional
relationship graph for efficient lookup.
"""
def __init__(self):
"""Initialize the manager by reading the Django setting and building the relationship graph."""
self._model_relationships = getattr(
settings,
'SYNC_MODEL_CHILDREN_SHARING',
{}
)
self._relationship_graph = {}
def _build_relationship_graph(self) -> Dict[str, List[List[str]]]:
"""
Build a bidirectional relationship graph.
Returns:
Dict[str, List[List[str]]]: Dictionary where keys are model names and values
are lists of relationship paths (arrays of relationship names).
"""
graph = {}
# Add direct relationships (original models to their children)
for model_name, relationships in self._model_relationships.items():
if model_name not in graph:
graph[model_name] = []
graph[model_name].append(relationships)
# Build reverse relationships (children back to original models)
for original_model_name, relationships in self._model_relationships.items():
try:
current_model = model_registry.get_model(original_model_name)
if current_model is None:
print(f'missing {original_model_name}')
continue
current_reverse_path = []
for relationship_name in relationships:
# Get the related model through _meta
try:
field = None
# Try to find the field in the model's _meta
for f in current_model._meta.get_fields():
if hasattr(f, 'related_name') and f.related_name == relationship_name:
field = f
break
elif hasattr(f, 'name') and f.name == relationship_name:
field = f
break
if field is None:
continue
# Get the related model
if hasattr(field, 'related_model'):
related_model = field.related_model
elif hasattr(field, 'model'):
related_model = field.model
else:
continue
related_model_name = related_model.__name__
# Find the reverse relationship name
reverse_relationship_name = self._find_reverse_relationship(
relationship_name, current_model, related_model
)
if reverse_relationship_name:
current_reverse_path.append(reverse_relationship_name)
# Add the reverse path
if related_model_name not in graph:
graph[related_model_name] = []
# The path back is just the reverse relationship name
graph[related_model_name] = current_reverse_path[::-1] # make a reverse copy
current_model = related_model
except Exception as e:
logger.info(f'error 2 > {e}')
# Skip problematic relationships
continue
except Exception as e:
logger.info(f'error > {e}')
continue
return graph
def _find_reverse_relationship(self, original_relationship_name, in_model, for_model):
"""
Find the reverse relationship name from from_model to to_model.
Args:
from_model: The model to search relationships from
to_model: The target model to find relationship to
original_relationship_name: The original relationship name for context
Returns:
str or None: The reverse relationship name if found
"""
print(f'reverse of {original_relationship_name} in = {in_model} for {for_model} ')
try:
for field in for_model._meta.get_fields():
# Check ForeignKey, OneToOneField fields
# print(f'{for_model} : field name = {field.name} / field.related_model = {field.related_model == in_model}')
if hasattr(field, 'related_model') and field.related_model == in_model:
return field.name
### possible improvements to do here if multiple relationships of the same type
# Check if this field has a related_name that matches our original relationship
# if hasattr(field, 'related_name') and field.related_name == original_relationship_name:
# # This is the reverse of our original relationship
# return field.name
# elif not hasattr(field, 'related_name') or field.related_name is None:
# default_name = f"{in_model._meta.related_query_name}"
# print(f'no related name: {default_name} / {original_relationship_name.rstrip('s')} ')
# if default_name == original_relationship_name.rstrip('s'): # Simple heuristic
# return field.name
# Check reverse relationships
if hasattr(field, 'field') and hasattr(field.field, 'model'):
if field.field.model == in_model:
if field.get_accessor_name() == original_relationship_name:
return field.field.name
except Exception as e:
print(f'!! ERROR = {e}')
pass
return None
def get_relationships(self, model_name: str) -> List[str]:
"""
Get the list of direct relationships for a given model name.
Args:
model_name (str): The name of the model to look up
Returns:
List[str]: List of relationship names for the model.
Returns empty list if model is not found.
"""
return self._model_relationships.get(model_name, [])
def get_relationship_paths(self, model_name: str) -> List[List[str]]:
"""
Get all relationship paths for a given model name.
This includes both direct relationships and reverse paths.
Args:
model_name (str): The name of the model to look up
Returns:
List[List[str]]: List of relationship paths (each path is a list of relationship names).
Returns empty list if model is not found.
"""
if not self._relationship_graph:
self._relationship_graph = self._build_relationship_graph()
logger.info(f'self._relationship_graph = {self._relationship_graph}')
return self._relationship_graph.get(model_name, [])
def get_relationship_graph(self) -> Dict[str, List[List[str]]]:
"""
Get the complete relationship graph.
Returns:
Dict[str, List[List[str]]]: The complete relationship graph
"""
return self._relationship_graph.copy()
# Create a singleton instance to use throughout the application
sync_model_manager = SyncModelChildrenManager()

23
sync/models/base.py
Unescape View File

@ -1,11 +1,10 @@
from django.db import models
from django.utils.timezone import now
from django.conf import settings
from django.apps import apps
from typing import List, Set
from ..model_manager import sync_model_manager
from collections import defaultdict
import logging
logger = logging.getLogger(__name__)
@ -41,8 +40,8 @@ class BaseModel(models.Model):
def update_data_access_list(self):
related_instances = self.sharing_related_instances()
data_access_ids = {instance.data_access_ids for instance in related_instances}
data_access_ids.update(self.data_access_ids)
data_access_ids = [instance.data_access_ids for instance in related_instances if isinstance(instance, BaseModel)]
data_access_ids.extend(self.data_access_ids)
self.data_access_ids = data_access_ids
# DataAccess = apps.get_model('sync', 'DataAccess')
@ -206,17 +205,21 @@ class BaseModel(models.Model):
return instances
def get_shared_children(self, processed_objects):
relationships_arrays = sync_model_manager.get_relationship_paths(self.__class__.__name__)
instances = []
sync_models = getattr(settings, 'SYNC_MODEL_CHILDREN_SHARING', {})
relationships = sync_models[self.__class__.__name__]
if relationships:
return self.get_shared_children_from_relationships(relationships, processed_objects)
if relationships_arrays:
for relationships in relationships_arrays:
children = self.get_shared_children_from_relationships(relationships, processed_objects)
instances.extend(children)
else:
return self.get_recursive_children(processed_objects)
instances.extend(self.get_recursive_children(processed_objects))
return instances
def get_shared_children_from_relationships(self, relationships, processed_objects):
# print(f'relationships = {relationships}')
print(f'>>> {self.__class__.__name__} : relationships = {relationships}')
current = [self]
for relationship in relationships:
# print(f'> relationship = {relationship}')

3
sync/models/data_access.py
Unescape View File

@ -69,8 +69,7 @@ class DataAccess(BaseModel):
if model_class:
try:
obj = model_class.objects.get(id=self.model_id)
related_instance = obj.sharing_related_instances()
related_instance = obj.sharing_related_instances() # here we want instances granted by the DataAccess, including SYNC_MODEL_CHILDREN_SHARING
related_instance.append(obj)
with transaction.atomic():

185
sync/registry.py
Unescape View File

@ -2,29 +2,24 @@ from django.conf import settings
from django.apps import apps
from django.contrib.auth import get_user_model
from .models import BaseModel
import threading
import logging
from typing import List, Optional, Dict
import importlib
logger = logging.getLogger(__name__)
User = get_user_model()
class ModelRegistry:
def __init__(self):
self._registry = {}
def load_sync_apps(self):
base_model = get_abstract_model_class('BaseModel')
sync_apps = getattr(settings, 'SYNC_APPS', {})
for app_label, config in sync_apps.items():
app_models = apps.get_app_config(app_label).get_models()
for model in app_models:
if hasattr(model, '_meta') and not model._meta.abstract:
if issubclass(model, BaseModel) or model == User:
if issubclass(model, base_model) or model == get_user_model():
model_name = model.__name__
if self.should_sync_model(model_name, config):
self.register(model)
@ -47,6 +42,10 @@ class ModelRegistry:
# Global instance
model_registry = ModelRegistry()
def get_abstract_model_class(model_name):
module = importlib.import_module('sync.models')
return getattr(module, model_name)
class DeviceRegistry:
"""Thread-safe registry to track device IDs associated with model instances."""
@ -113,173 +112,3 @@ class RelatedUsersRegistry:
# Global instance
related_users_registry = RelatedUsersRegistry()
class SyncModelChildrenManager:
"""
Manager class for handling model children sharing configuration.
Reads the SYNC_MODEL_CHILDREN_SHARING setting once and builds a bidirectional
relationship graph for efficient lookup.
"""
def __init__(self):
"""Initialize the manager by reading the Django setting and building the relationship graph."""
self._model_relationships = getattr(
settings,
'SYNC_MODEL_CHILDREN_SHARING',
{}
)
self._relationship_graph = self._build_relationship_graph()
logger.info(f'self._relationship_graph = {self._relationship_graph}')
def _build_relationship_graph(self) -> Dict[str, List[List[str]]]:
"""
Build a bidirectional relationship graph.
Returns:
Dict[str, List[List[str]]]: Dictionary where keys are model names and values
are lists of relationship paths (arrays of relationship names).
"""
graph = {}
# Add direct relationships (original models to their children)
for model_name, relationships in self._model_relationships.items():
if model_name not in graph:
graph[model_name] = []
# Add direct relationships as single-item arrays
for relationship in relationships:
graph[model_name].append([relationship])
# Build reverse relationships (children back to original models)
for original_model_name, relationships in self._model_relationships.items():
try:
original_model = model_registry.get_model(original_model_name)
if original_model is None:
continue
for relationship_name in relationships:
# Get the related model through _meta
try:
field = None
# Try to find the field in the model's _meta
for f in original_model._meta.get_fields():
if hasattr(f, 'related_name') and f.related_name == relationship_name:
field = f
break
elif hasattr(f, 'name') and f.name == relationship_name:
field = f
break
if field is None:
continue
# Get the related model
if hasattr(field, 'related_model'):
related_model = field.related_model
elif hasattr(field, 'model'):
related_model = field.model
else:
continue
related_model_name = related_model.__name__
# Find the reverse relationship name
reverse_relationship_name = self._find_reverse_relationship(
related_model, original_model, relationship_name
)
if reverse_relationship_name:
# Add the reverse path
if related_model_name not in graph:
graph[related_model_name] = []
# The path back is just the reverse relationship name
graph[related_model_name].append([reverse_relationship_name])
except Exception as e:
# Skip problematic relationships
continue
except Exception as e:
# Skip problematic models
continue
return graph
def _find_reverse_relationship(self, from_model, to_model, original_relationship_name):
"""
Find the reverse relationship name from from_model to to_model.
Args:
from_model: The model to search relationships from
to_model: The target model to find relationship to
original_relationship_name: The original relationship name for context
Returns:
str or None: The reverse relationship name if found
"""
try:
for field in from_model._meta.get_fields():
# Check ForeignKey, OneToOneField fields
if hasattr(field, 'related_model') and field.related_model == to_model:
# Check if this field has a related_name that matches our original relationship
if hasattr(field, 'related_name') and field.related_name == original_relationship_name:
# This is the reverse of our original relationship
return field.name
elif not hasattr(field, 'related_name') or field.related_name is None:
# Default reverse relationship name
default_name = f"{to_model._meta.model_name}"
if default_name == original_relationship_name.rstrip('s'): # Simple heuristic
return field.name
# Check reverse relationships
if hasattr(field, 'field') and hasattr(field.field, 'model'):
if field.field.model == to_model:
if field.get_accessor_name() == original_relationship_name:
return field.field.name
except Exception:
pass
return None
def get_relationships(self, model_name: str) -> List[str]:
"""
Get the list of direct relationships for a given model name.
Args:
model_name (str): The name of the model to look up
Returns:
List[str]: List of relationship names for the model.
Returns empty list if model is not found.
"""
return self._model_relationships.get(model_name, [])
def get_relationship_paths(self, model_name: str) -> List[List[str]]:
"""
Get all relationship paths for a given model name.
This includes both direct relationships and reverse paths.
Args:
model_name (str): The name of the model to look up
Returns:
List[List[str]]: List of relationship paths (each path is a list of relationship names).
Returns empty list if model is not found.
"""
return self._relationship_graph.get(model_name, [])
def get_relationship_graph(self) -> Dict[str, List[List[str]]]:
"""
Get the complete relationship graph.
Returns:
Dict[str, List[List[str]]]: The complete relationship graph
"""
return self._relationship_graph.copy()
# Create a singleton instance to use throughout the application
sync_model_manager = SyncModelChildrenManager()

2
tournaments/models/event.py
Unescape View File

@ -26,7 +26,7 @@ class Event(BaseModel):
def save(self, *args, **kwargs):
if self.creator:
self.creator_full_name = self.creator.full_name()
super(Event, self).save(*args, **kwargs)
super().save(*args, **kwargs)
def display_name(self):
if self.club and self.club.name and self.name:

Write Preview
Loading…
Cancel
Save