Coverage for .nox/test-3-9/lib/python3.9/site-packages/nskit/mixer/components/filesystem_object.py: 99%

1"""Base Filesystem object and template string.""" 

2from abc import ABC, abstractmethod 

3from pathlib import Path 

4from typing import Any, Callable, Dict, Optional, Union 

5 

6from pydantic import Field, GetCoreSchemaHandler 

7from pydantic_core import core_schema, CoreSchema 

8 

9from nskit.common.configuration import BaseConfiguration 

10from nskit.mixer.utilities import JINJA_ENVIRONMENT_FACTORY 

11 

12 

13class TemplateStr(str): 

14 """Type that includes jinja templating.""" 

15 

16 @classmethod 

17 def __get_pydantic_core_schema__( 

18 cls, source_type: Any, handler: GetCoreSchemaHandler # noqa: U100 

19 ) -> CoreSchema: 

20 """Get the schema.""" 

21 return core_schema.no_info_after_validator_function(cls._validate_template_str, handler(str)) 

22 

23 @classmethod 

24 def _validate_template_str(cls, value: str): 

25 if ('{{' in value and '}}' in value) or ('{%' in value and '%}' in value): 

26 return cls(value) 

27 else: 

28 raise ValueError('Template str needs jinja syntax.') 

29 

30 def __call__(self, context: Optional[Dict[str, Union[str, int]]] = None): 

31 """Render the template.""" 

32 if context is None: 

33 context = {} 

34 return JINJA_ENVIRONMENT_FACTORY.environment.from_string(self).render(context) 

35 

36 

37class FileSystemObject(ABC, BaseConfiguration): 

38 """Abstract pydantic model that acts as the base for filesystem objects.""" 

39 

40 id_: Optional[Union[int, str]] = Field(None, description="An Id to refer to the object when e.g. the name is a template string or callable") 

41 name: Optional[Union[TemplateStr, str, Callable]] = Field(..., validate_default=True, description='The name of the filesystem object, can be a string, TemplateStr or callable (which returns a string)') 

42 

43 def render_name(self, context: Optional[Dict[str, Union[str, int]]] = None): 

44 """Render the name if it is a template string or callable.""" 

45 if context is None: 

46 context = {} 

47 if not isinstance(context, dict): 

48 raise TypeError(f'Context must be a dict, not {type(context)}') 

49 if isinstance(self.name, TemplateStr) or not isinstance(self.name, str): 

50 # Either a callable or TemplateStr (which is a callable) 

51 rendered_name = self.name(context) 

52 else: 

53 rendered_name = self.name 

54 return rendered_name 

55 

56 def _repr(self, context=None, **kwargs): # noqa: U100 

57 if isinstance(self.name, TemplateStr): 

58 name_value = self.name 

59 name = 'name <TemplateStr>' 

60 elif isinstance(self.name, str): 

61 name = 'name' 

62 name_value = self.name 

63 else: 

64 name_value = '<callable>' 

65 name = 'name' 

66 rendered_name = name_value 

67 if context: 

68 rendered_name = self.render_name(context) 

69 if self.id_: 

70 id_ = f'id: {self.id_}, ' 

71 else: 

72 id_ = '' 

73 return f'{rendered_name} = {self.__class__.__name__}({id_}{name}: {name_value})' 

74 

75 def __repr__(self): 

76 """repr(x).""" 

77 return self._repr() 

78 

79 def get_path(self, base_path: Path, context: Dict[str, Union[str, int]], override_path: Optional[Path] = None): 

80 """Get the object path. Can be overriden with the override_path (relative to the base path).""" 

81 path = None 

82 if override_path: 

83 path = Path(base_path)/Path(override_path) 

84 else: 

85 rendered_name = self.render_name(context) 

86 if rendered_name is not None: 

87 path = Path(base_path) / rendered_name 

88 return path 

89 

90 @abstractmethod 

91 def write(self, base_path: Path, context: Dict[str, Any], override_path: Optional[Path] = None): 

92 """Write the object to the appropriate path within the ``base_path``.""" 

93 raise NotImplementedError() 

94 

95 @abstractmethod 

96 def dryrun(self, base_path: Path, context: Dict[str, Any], override_path: Optional[Path] = None): 

97 """Preview the file contents using the context.""" 

98 raise NotImplementedError() 

99 

100 @abstractmethod 

101 def validate(self, base_path: Path, context: Dict[str, Any], override_path: Optional[Path] = None): 

102 """Validate the output against expected.""" 

103 raise NotImplementedError()