Coverage for /opt/conda/envs/apienv/lib/python3.10/site-packages/daiquiri/core/components/params.py: 70%

1#!/usr/bin/env python 

2# -*- coding: utf-8 -*- 

3from abc import ABC 

4from marshmallow import Schema 

5 

6import logging 

7 

8logger = logging.getLogger(__name__) 

9 

10 

11class ParamHandler(ABC): 

12 """Parameter Handler 

13 

14 Handler for an individual parameter, two methods should be implemented 

15 in the child, before, and after. These will be called before an actor is 

16 started and after an actor finishes. 

17 

18 A handler can also store an arbitrary value so that for example a position 

19 can be restored after an actor finishes 

20 """ 

21 

22 def __init__(self): 

23 self._value = None 

24 

25 def set_cache(self, value): 

26 self._value = value 

27 

28 def get_cache(self): 

29 return self._value 

30 

31 def do_before(self, value): 

32 logger.info(f"Running before {self.__class__.__name__}") 

33 self.before(value) 

34 logger.info(f"Before {self.__class__.__name__} finished") 

35 

36 def do_after(self, value): 

37 logger.info(f"Running after {self.__class__.__name__}") 

38 self.after(value) 

39 logger.info(f"After {self.__class__.__name__} finished") 

40 

41 def before(self, value): 

42 """Implemented in child""" 

43 pass 

44 

45 def after(self, value): 

46 """Implemented in child""" 

47 pass 

48 

49 

50class ParamsHandlerMeta(type): 

51 """Parameters Handler Metaclass 

52 

53 This meta class pops attributes off the child instance into a handlers 

54 attribute so classes can be declared in a simple fashion 

55 """ 

56 

57 def __new__(mcs, name, bases, attrs): 

58 handlers = {} 

59 to_rem = [] 

60 for a, v in attrs.items(): 

61 if isinstance(v, ParamHandler): 

62 handlers[a] = v 

63 to_rem.append(a) 

64 

65 for r in to_rem: 

66 del attrs[r] 

67 

68 klass = super().__new__(mcs, name, bases, attrs) 

69 klass.handlers = handlers 

70 

71 return klass 

72 

73 

74class ParamsHandler(metaclass=ParamsHandlerMeta): 

75 """Parameters Handler 

76 

77 This class will find the parameter instance to maniplate from the 

78 registered parameters. 

79 """ 

80 

81 def handle(self, params, before=True): 

82 for p, v in params.items(): 

83 if p in self.handlers: 

84 meth = getattr(self.handlers[p], "do_before" if before else "do_after") 

85 meth(v) 

86 

87 

88class ParamSchema(Schema): 

89 """Parameteres Schema 

90 

91 A special marshmallow schema that has `ParamsHandler` attribute in 

92 addition to the schema. These handlers will be used if a schema contains 

93 values from this ParamSchema 

94 """ 

95 

96 handler = None 

97 

98 def __init__(self, *args, **kwargs): 

99 super().__init__(*args, **kwargs) 

100 

101 self.Meta.cache = True 

102 

103 if self.handler is not None: 

104 if not isinstance(self.handler, ParamsHandler): 

105 raise AttributeError( 

106 "Parameters handler is not an instance of <ParamsHandler>" 

107 )