Coverage for gws-app/gws/base/model/manager.py: 71%

48 statements  

« prev     ^ index     » next       coverage.py v7.16.2, created at 2026-10-05 13:35 +0200

1"""Model manager.""" 

2 

3from typing import Optional, cast 

4 

5import gws 

6from . import default_model 

7 

8 

9class Object(gws.ModelManager): 

10 """Model manager.""" 

11 

12 def get_model(self, uid, user=None, access=None): 

13 model = cast(gws.Model, self.root.get(uid, gws.ext.object.model)) 

14 if not model: 

15 return 

16 if user and access and not user.can(access, model): 

17 return 

18 return model 

19 

20 def find_model(self, *objects, user=None, access=None): 

21 for obj in objects: 

22 if not obj: 

23 continue 

24 p = self._find(obj, user, access) 

25 if p: 

26 return p 

27 

28 def _find(self, obj, user, access): 

29 """Return the first model of an object the user can access.""" 

30 for model in getattr(obj, 'models', []): 

31 if user and access and not user.can(access, model): 

32 continue 

33 return model 

34 

35 def editable_models(self, project, user): 

36 res = {} 

37 

38 def _collect(obj): 

39 for model in getattr(obj, 'models', []): 

40 if model.isEditable and user.can_edit(model): 

41 res[model.uid] = model 

42 

43 if project.map: 

44 for la in project.map.rootLayer.find_descendants(gws.ext.object.layer): 

45 _collect(la) 

46 

47 _collect(project) 

48 _collect(self.root.app) 

49 

50 queue = [] 

51 for model in res.values(): 

52 queue.extend(model.related_models()) 

53 

54 while queue: 

55 model = queue.pop(0) 

56 if model.uid in res: 

57 continue 

58 if not user.can_read(model): 

59 continue 

60 res[model.uid] = model 

61 queue.extend(model.related_models()) 

62 

63 return sorted(res.values(), key=lambda m: m.title) 

64 

65 def default_model(self): 

66 return self.root.create_shared(default_model.Object, uid='gws.base.core.default_model.Object')