Coverage for .nox/test-3-9/lib/python3.9/site-packages/nskit/mixer/components/filesystem_object.py: 99%
67 statements
« prev ^ index » next coverage.py v7.4.2, created at 2024-02-25 17:38 +0000
« prev ^ index » next coverage.py v7.4.2, created at 2024-02-25 17:38 +0000
1"""Base Filesystem object and template string."""
2from abc import ABC, abstractmethod
3from pathlib import Path
4from typing import Any, Callable, Dict, Optional, Union
6from pydantic import Field, GetCoreSchemaHandler
7from pydantic_core import core_schema, CoreSchema
9from nskit.common.configuration import BaseConfiguration
10from nskit.mixer.utilities import JINJA_ENVIRONMENT_FACTORY
13class TemplateStr(str):
14 """Type that includes jinja templating."""
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))
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.')
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)
37class FileSystemObject(ABC, BaseConfiguration):
38 """Abstract pydantic model that acts as the base for filesystem objects."""
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)')
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
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})'
75 def __repr__(self):
76 """repr(x)."""
77 return self._repr()
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
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()
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()
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()