Coverage for gws-app/gws/base/application/middleware.py: 95%

1"""Middleware manager.""" 

2 

3import gws 

4 

5 

6class Object(gws.MiddlewareManager): 

7 """Middleware manager.""" 

8 

9 objectMap: dict[str, gws.Node] 

10 """Registered objects by name.""" 

11 deps: dict[str, list[str]] 

12 """Dependency names by object name.""" 

13 names: list[str] 

14 """Object names in dependency order.""" 

15 

16 def __init__(self): 

17 self.objectMap = {} 

18 self.deps = {} 

19 self.names = [] 

20 self.sorted = False 

21 

22 def register(self, obj, name, depends_on=None): 

23 self.objectMap[name] = obj 

24 self.deps[name] = depends_on 

25 self.sorted = False 

26 

27 def objects(self): 

28 if not self.sorted: 

29 self._sort() 

30 self.sorted = True 

31 return [self.objectMap[name] for name in self.names] 

32 

33 def _sort(self): 

34 """Sort the object names in dependency order.""" 

35 self.names = [] 

36 colors = {} 

37 for name in self.objectMap: 

38 self._sort_visit(name, colors, []) 

39 

40 def _sort_visit(self, name, colors, stack): 

41 """Visit a name and its dependencies depth first, raise on cycles and unknown names.""" 

42 stack = stack + [name] 

43 

44 if colors.get(name) == 2: 

45 return 

46 if colors.get(name) == 1: 

47 raise gws.Error('middleware: cyclic dependency: ' + '->'.join(stack)) 

48 

49 if name not in self.objectMap: 

50 raise gws.Error('middleware: not found: ' + '->'.join(stack)) 

51 

52 colors[name] = 1 

53 

54 depends_on = self.deps[name] 

55 if depends_on: 

56 for d in depends_on: 

57 self._sort_visit(d, colors, stack) 

58 

59 colors[name] = 2 

60 self.names.append(name)