Skip to content

nskit.mixer

nskit.mixer.

Building blocks to build up repo templates and mix (instantiate) them.

Components

nskit.mixer.components.file.File

Bases: FileSystemObject

File component.

Source code in src/nskit/mixer/components/file.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
class File(FileSystemObject):
    """File component."""

    content: Union[Resource, str, bytes, Path, Callable] = Field("", description="The file content")

    def render_content(self, context: dict[str, Any]):  # pylint: disable=arguments-differ
        """Return the rendered content using the context and the Jinja environment."""
        if context is None:
            context = {}
        if isinstance(self.content, Resource):
            content = self.content.load()
        elif isinstance(self.content, Path):
            with open(self.content) as f:
                content = f.read()
        elif isinstance(self.content, Callable):
            content = self.content(context)
        else:
            content = self.content
        if isinstance(content, str):
            # If it is a string, we render the content
            content = JINJA_ENVIRONMENT_FACTORY.environment.from_string(content).render(**context)
        return content

    def write(self, base_path: Path, context: dict[str, Any], override_path: Optional[Path] = None):
        """Write the rendered content to the appropriate path within the ``base_path``."""
        file_path = self.get_path(base_path, context, override_path)
        content = self.render_content(context)
        response = {}
        if content is not None:
            if isinstance(content, str):
                open_str = "w"
            elif isinstance(content, bytes):
                open_str = "wb"
            with file_path.open(open_str) as output_file:
                output_file.write(content)
            response[file_path] = content
        return response

    def dryrun(self, base_path: Path, context: dict[str, Any], override_path: Optional[Path] = None):
        """Preview the file contents using the context."""
        file_path = self.get_path(base_path, context, override_path)
        content = self.render_content(context)
        result = {}
        if content is not None:
            result[file_path] = content
        return result

    def validate(self, base_path: Path, context: dict[str, Any], override_path: Optional[Path] = None):
        """Validate the output against expected."""
        missing = []
        errors = []
        ok = []
        path = self.get_path(base_path, context, override_path)
        content = self.render_content(context)
        if content is not None:
            if not path.exists():
                missing.append(path)
            else:
                if isinstance(content, bytes):
                    read_str = "rb"
                else:
                    read_str = "r"
                with open(path, read_str) as f:
                    if f.read() != self.render_content(context):
                        errors.append(path)
                    else:
                        ok.append(path)
        return missing, errors, ok

dryrun(base_path, context, override_path=None)

Preview the file contents using the context.

Source code in src/nskit/mixer/components/file.py 
50
51
52
53
54
55
56
57
def dryrun(self, base_path: Path, context: dict[str, Any], override_path: Optional[Path] = None):
    """Preview the file contents using the context."""
    file_path = self.get_path(base_path, context, override_path)
    content = self.render_content(context)
    result = {}
    if content is not None:
        result[file_path] = content
    return result

render_content(context)

Return the rendered content using the context and the Jinja environment.

Source code in src/nskit/mixer/components/file.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def render_content(self, context: dict[str, Any]):  # pylint: disable=arguments-differ
    """Return the rendered content using the context and the Jinja environment."""
    if context is None:
        context = {}
    if isinstance(self.content, Resource):
        content = self.content.load()
    elif isinstance(self.content, Path):
        with open(self.content) as f:
            content = f.read()
    elif isinstance(self.content, Callable):
        content = self.content(context)
    else:
        content = self.content
    if isinstance(content, str):
        # If it is a string, we render the content
        content = JINJA_ENVIRONMENT_FACTORY.environment.from_string(content).render(**context)
    return content

validate(base_path, context, override_path=None)

Validate the output against expected.

Source code in src/nskit/mixer/components/file.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def validate(self, base_path: Path, context: dict[str, Any], override_path: Optional[Path] = None):
    """Validate the output against expected."""
    missing = []
    errors = []
    ok = []
    path = self.get_path(base_path, context, override_path)
    content = self.render_content(context)
    if content is not None:
        if not path.exists():
            missing.append(path)
        else:
            if isinstance(content, bytes):
                read_str = "rb"
            else:
                read_str = "r"
            with open(path, read_str) as f:
                if f.read() != self.render_content(context):
                    errors.append(path)
                else:
                    ok.append(path)
    return missing, errors, ok

write(base_path, context, override_path=None)

Write the rendered content to the appropriate path within the base_path.

Source code in src/nskit/mixer/components/file.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def write(self, base_path: Path, context: dict[str, Any], override_path: Optional[Path] = None):
    """Write the rendered content to the appropriate path within the ``base_path``."""
    file_path = self.get_path(base_path, context, override_path)
    content = self.render_content(context)
    response = {}
    if content is not None:
        if isinstance(content, str):
            open_str = "w"
        elif isinstance(content, bytes):
            open_str = "wb"
        with file_path.open(open_str) as output_file:
            output_file.write(content)
        response[file_path] = content
    return response

nskit.mixer.components.folder.Folder

Bases: FileSystemObject

Folder component.

Source code in src/nskit/mixer/components/folder.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
class Folder(FileSystemObject):
    """Folder component."""

    contents: list[Union[File, "Folder"]] = Field(default_factory=list, description="The folder contents")

    def write(self, base_path: Path, context: dict[str, Any], override_path: Optional[Path] = None):
        """Write the rendered content to the appropriate path within the ``base_path``."""
        folder_path = self.get_path(base_path, context, override_path)
        folder_path.mkdir(exist_ok=True, parents=True)
        contents_dict = {}
        for obj in self.contents:
            contents_dict.update(obj.write(folder_path, context))
        return {folder_path: contents_dict}

    def dryrun(self, base_path: Path, context: dict[str, Any], override_path: Optional[Path] = None):
        """Preview the file contents using the context."""
        folder_path = self.get_path(base_path, context, override_path)
        contents_dict = {}
        for u in self.contents:
            contents_dict.update(u.dryrun(folder_path, context))
        result = {folder_path: contents_dict}
        return result

    def validate(self, base_path: Path, context: dict[str, Any], override_path: Optional[Path] = None):
        """Validate the output against expected."""
        missing = []
        errors = []
        ok = []
        path = self.get_path(base_path, context, override_path)
        if not path.exists():
            missing.append(path)
        for child in self.contents:
            child_missing, child_errors, child_ok = child.validate(path, context)
            missing += child_missing
            errors += child_errors
            ok += child_ok
        if not missing and not errors:
            ok.append(path)
        return missing, errors, ok

    @field_validator("contents", mode="before")
    @classmethod
    def _validate_contents_ids_unique(cls, contents):
        if contents:
            ids_ = []
            for item in contents:
                id_ = None
                if isinstance(item, FileSystemObject):
                    id_ = item.id_
                if isinstance(item, dict):
                    id_ = item.get("id_", None)
                if id_ is None:
                    # No id_ provided
                    continue
                if id_ in ids_:
                    raise ValueError(
                        f"IDs for contents must be unique. The ID({id_}) already exists in the folder contents"
                    )
                ids_.append(id_)
        return contents

    def index(self, name_or_id):
        """Get the index of a specific file or folder given the name (or ID)."""
        for i, item in enumerate(self.contents):
            if item.id_ == name_or_id or item.name == name_or_id:
                return i
        raise KeyError(f"Name or id_ {name_or_id} not found in contents")

    def __getitem__(self, name_or_id):
        """Get the item by name or id."""
        index = self.index(name_or_id)
        return self.contents[index]

    def __setitem__(self, name_or_id, value):
        """Set an item by name or id."""
        try:
            index = self.index(name_or_id)
            self.contents.pop(index)
            self.contents.insert(index, value)
        except KeyError:
            self.contents.append(value)

    def _repr(self, context=None, indent=0, **kwargs):  # noqa: U100
        """Represent the contents of the folder."""
        indent_ = " " * indent
        line_start = f"\n{indent_}|- "
        contents_repr = ""
        if self.contents:
            contents = sorted(self.contents, key=lambda x: isinstance(x, Folder))
            lines = [u._repr(context=context, indent=indent + 2) for u in contents]
            contents_repr = ":" + line_start.join([""] + lines)
        return f"{super()._repr(context=context)}{contents_repr}"

__getitem__(name_or_id)

Get the item by name or id.

Source code in src/nskit/mixer/components/folder.py 
80
81
82
83
def __getitem__(self, name_or_id):
    """Get the item by name or id."""
    index = self.index(name_or_id)
    return self.contents[index]

__setitem__(name_or_id, value)

Set an item by name or id.

Source code in src/nskit/mixer/components/folder.py 
85
86
87
88
89
90
91
92
def __setitem__(self, name_or_id, value):
    """Set an item by name or id."""
    try:
        index = self.index(name_or_id)
        self.contents.pop(index)
        self.contents.insert(index, value)
    except KeyError:
        self.contents.append(value)

dryrun(base_path, context, override_path=None)

Preview the file contents using the context.

Source code in src/nskit/mixer/components/folder.py 
26
27
28
29
30
31
32
33
def dryrun(self, base_path: Path, context: dict[str, Any], override_path: Optional[Path] = None):
    """Preview the file contents using the context."""
    folder_path = self.get_path(base_path, context, override_path)
    contents_dict = {}
    for u in self.contents:
        contents_dict.update(u.dryrun(folder_path, context))
    result = {folder_path: contents_dict}
    return result

index(name_or_id)

Get the index of a specific file or folder given the name (or ID).

Source code in src/nskit/mixer/components/folder.py 
73
74
75
76
77
78
def index(self, name_or_id):
    """Get the index of a specific file or folder given the name (or ID)."""
    for i, item in enumerate(self.contents):
        if item.id_ == name_or_id or item.name == name_or_id:
            return i
    raise KeyError(f"Name or id_ {name_or_id} not found in contents")

validate(base_path, context, override_path=None)

Validate the output against expected.

Source code in src/nskit/mixer/components/folder.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def validate(self, base_path: Path, context: dict[str, Any], override_path: Optional[Path] = None):
    """Validate the output against expected."""
    missing = []
    errors = []
    ok = []
    path = self.get_path(base_path, context, override_path)
    if not path.exists():
        missing.append(path)
    for child in self.contents:
        child_missing, child_errors, child_ok = child.validate(path, context)
        missing += child_missing
        errors += child_errors
        ok += child_ok
    if not missing and not errors:
        ok.append(path)
    return missing, errors, ok

write(base_path, context, override_path=None)

Write the rendered content to the appropriate path within the base_path.

Source code in src/nskit/mixer/components/folder.py 
17
18
19
20
21
22
23
24
def write(self, base_path: Path, context: dict[str, Any], override_path: Optional[Path] = None):
    """Write the rendered content to the appropriate path within the ``base_path``."""
    folder_path = self.get_path(base_path, context, override_path)
    folder_path.mkdir(exist_ok=True, parents=True)
    contents_dict = {}
    for obj in self.contents:
        contents_dict.update(obj.write(folder_path, context))
    return {folder_path: contents_dict}

nskit.mixer.components.recipe.Recipe

Bases: Folder

The base Recipe object.

A Recipe is a folder, with additional methods for handling context for the Jinja templating. It also includes hooks that can be run before rendering (pre-hooks) e.g. checking or changing values, and after (post-hooks) e.g. running post-generation steps.

Source code in src/nskit/mixer/components/recipe.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
class Recipe(Folder):
    """The base Recipe object.

    A Recipe is a folder, with additional methods for handling context for the Jinja templating.
    It also includes hooks that can be run before rendering (``pre-hooks``) e.g. checking or changing values,
    and after (``post-hooks``) e.g. running post-generation steps.
    """

    name: str = Field(None, validate_default=True, description="The repository name")
    version: Optional[str] = Field(None, description="The recipe version")  # type: ignore
    pre_hooks: list[Hook] = Field(
        default_factory=list,
        validate_default=True,
        description="Hooks that can be used to modify a recipe path and context before writing",
    )
    post_hooks: list[Hook] = Field(
        default_factory=list,
        validate_default=True,
        description="Hooks that can be used to modify a recipe path and context after writing",
    )
    extension_name: Optional[str] = Field(None, description="The name of the recipe as an extension to load.")

    # Config path constants (can be overridden by subclasses)
    config_dir: ClassVar[str] = ".recipe"
    config_filename: ClassVar[str] = "config.yml"

    @property
    def recipe(self):
        """Recipe context."""
        extension_name = self.extension_name
        if extension_name is None:
            extension_name = self.__class__.__name__
        return {
            "name": f"{self.__class__.__module__}:{self.__class__.__name__}",
            "version": self.version,
            "extension_name": extension_name,
        }

    def create(self, base_path: Optional[Path] = None, override_path: Optional[Path] = None, **additional_context):
        """Create the recipe.

        Use the configured parameters and any additional context as kwargs to create the recipe at the
        base path (or current directory if not provided).
        """
        if base_path is None:
            base_path = Path.cwd()
        else:
            base_path = Path(base_path)
        context = self.context
        context.update(additional_context)
        recipe_path = self.get_path(base_path, context, override_path=override_path)
        for hook in self.pre_hooks:
            recipe_path, context = hook(recipe_path, context, recipe=self)
        content = self.write(recipe_path.parent, context, override_path=recipe_path.name)
        recipe_path = next(iter(content.keys()))
        for hook in self.post_hooks:
            recipe_path, context = hook(recipe_path, context, recipe=self)
        self._write_batch(Path(recipe_path))
        return {Path(recipe_path): next(iter(content.values()))}

    def _write_batch(self, folder_path: Path):
        """Write out the parameters used.

        When we use this we want to keep track of what parameters were used to enable rerunning.
        This methods writes this into the generated folder as a YAML file.
        """
        batch_path = Path(folder_path) / ".recipe-batch.yaml"
        if batch_path.exists():
            with batch_path.open() as f:
                batch = yaml.loads(f.read())
        else:
            batch = []
        batch.append(self.recipe_batch)
        with batch_path.open("w") as f:
            f.write(yaml.dumps(batch))

    @property
    def recipe_batch(self):
        """Get information about the specific info of this recipe."""
        if sys.version_info.major <= 3 and sys.version_info.minor < 11:
            creation_time = dt.datetime.now().astimezone().isoformat()
        else:
            creation_time = dt.datetime.now(dt.UTC).isoformat()
        return {
            "context": self.__dump_context(ser=True),
            "nskit_version": __version__,
            "creation_time": creation_time,
            "recipe": self.recipe,
        }

    @property
    def context(self):
        """Get the context on the initialised recipe."""
        # This inherits (via FileSystemObject) from nskit.common.configuration:BaseConfiguration, which includes properties in model dumps
        return self.__dump_context()

    def __dump_context(self, ser=False):
        # Make sure it is serialisable if required
        if ser:
            mode = "json"
        else:
            mode = "python"
        context = self.model_dump(
            mode=mode,
            exclude={
                "context",
                "contents",
                "name",
                "id_",
                "post_hooks",
                "pre_hooks",
                "version",
                "recipe_batch",
                "recipe",
                "extension_name",
            },
        )
        context.update({"recipe": self.recipe})
        return context

    def __repr__(self):
        """Repr(x) == x.__repr__."""
        context = self.context
        return f"{self._repr(context=context)}\n\nContext: {context}"

    def dryrun(self, base_path: Optional[Path] = None, override_path: Optional[Path] = None, **additional_context):
        """See the recipe as a dry run."""
        combined_context = self.context
        combined_context.update(additional_context)
        if base_path is None:
            base_path = Path.cwd()
        return super().dryrun(base_path=base_path, context=combined_context, override_path=override_path)

    def validate(self, base_path: Optional[Path] = None, override_path: Optional[Path] = None, **additional_context):
        """Validate the created repo."""
        combined_context = self.context
        combined_context.update(additional_context)
        if base_path is None:
            base_path = Path.cwd()
        return super().validate(base_path=base_path, context=combined_context, override_path=override_path)

    @staticmethod
    def load(recipe_name: str, entrypoint: Optional[str] = None, initialize: bool = True, **kwargs):
        """Load a recipe as an extension.

        Args:
            recipe_name: Name of the recipe to load
            entrypoint: Recipe entrypoint to use (defaults to RECIPE_ENTRYPOINT)
            initialize: Whether to initialize the recipe instance (default True)
            **kwargs: Arguments to pass to recipe initialization

        Returns:
            Recipe instance if initialize=True, otherwise recipe class
        """
        if entrypoint is None:
            entrypoint = RECIPE_ENTRYPOINT

        recipe_klass = load_extension(entrypoint, recipe_name)
        if recipe_klass is None:
            raise ValueError(
                f"Recipe {recipe_name} not found, it may be mis-spelt or not installed. Available recipes: {get_extension_names(entrypoint)}"
            )

        if not initialize:
            recipe_klass.extension_name = recipe_name
            return recipe_klass

        recipe = recipe_klass(**kwargs)
        recipe.extension_name = recipe_name
        return recipe

    @staticmethod
    def inspect(
        recipe_name: str,
        entrypoint: Optional[str] = None,
        include_private: bool = False,
        include_folder: bool = False,
        include_base: bool = False,
    ):
        """Get the fields on a recipe as an extension.

        Args:
            recipe_name: Name of the recipe to inspect
            entrypoint: Recipe entrypoint to use (defaults to RECIPE_ENTRYPOINT)
            include_private: Include private fields
            include_folder: Include folder fields
            include_base: Include base recipe fields

        Returns:
            Signature of the recipe
        """
        if entrypoint is None:
            entrypoint = RECIPE_ENTRYPOINT

        recipe_klass = load_extension(entrypoint, recipe_name)
        if recipe_klass is None:
            raise ValueError(
                f"Recipe {recipe_name} not found, it may be mis-spelt or not installed. Available recipes: {get_extension_names(entrypoint)}"
            )
        sig = Recipe._inspect_basemodel(recipe_klass, include_private=include_private)
        if not include_folder:
            folder_sig = inspect.signature(Folder)
            params = [v for u, v in sig.parameters.items() if u not in folder_sig.parameters.keys() or u == "name"]
            sig = sig.replace(parameters=params)
        if not include_base:
            recipe_sig = inspect.signature(Recipe)
            params = [v for u, v in sig.parameters.items() if u not in recipe_sig.parameters.keys() or u == "name"]
            sig = sig.replace(parameters=params)
        return sig

    @staticmethod
    def _inspect_basemodel(kls, include_private: bool = False):
        sig = inspect.signature(kls)
        # we need to drop the private params
        params = []
        for u, v in sig.parameters.items():
            if not include_private and u.startswith("_"):
                continue
            if isinstance(v.annotation, type) and issubclass(v.annotation, BaseModel):
                params.append(
                    v.replace(default=Recipe._inspect_basemodel(v.annotation, include_private=include_private))
                )
            else:
                params.append(v)
        return sig.replace(parameters=params, return_annotation=kls)

name = Field(None, validate_default=True, description='The repository name') class-attribute instance-attribute

version = Field(None, description='The recipe version') class-attribute instance-attribute

contents = Field(default_factory=list, description='The folder contents') class-attribute instance-attribute

pre_hooks = Field(default_factory=list, validate_default=True, description='Hooks that can be used to modify a recipe path and context before writing') class-attribute instance-attribute

post_hooks = Field(default_factory=list, validate_default=True, description='Hooks that can be used to modify a recipe path and context after writing') class-attribute instance-attribute

extension_name = Field(None, description='The name of the recipe as an extension to load.') class-attribute instance-attribute

context property

Get the context on the initialised recipe.

create(base_path=None, override_path=None, **additional_context)

Create the recipe.

Use the configured parameters and any additional context as kwargs to create the recipe at the base path (or current directory if not provided).

Source code in src/nskit/mixer/components/recipe.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def create(self, base_path: Optional[Path] = None, override_path: Optional[Path] = None, **additional_context):
    """Create the recipe.

    Use the configured parameters and any additional context as kwargs to create the recipe at the
    base path (or current directory if not provided).
    """
    if base_path is None:
        base_path = Path.cwd()
    else:
        base_path = Path(base_path)
    context = self.context
    context.update(additional_context)
    recipe_path = self.get_path(base_path, context, override_path=override_path)
    for hook in self.pre_hooks:
        recipe_path, context = hook(recipe_path, context, recipe=self)
    content = self.write(recipe_path.parent, context, override_path=recipe_path.name)
    recipe_path = next(iter(content.keys()))
    for hook in self.post_hooks:
        recipe_path, context = hook(recipe_path, context, recipe=self)
    self._write_batch(Path(recipe_path))
    return {Path(recipe_path): next(iter(content.values()))}

dryrun(base_path=None, override_path=None, **additional_context)

See the recipe as a dry run.

Source code in src/nskit/mixer/components/recipe.py 
219
220
221
222
223
224
225
def dryrun(self, base_path: Optional[Path] = None, override_path: Optional[Path] = None, **additional_context):
    """See the recipe as a dry run."""
    combined_context = self.context
    combined_context.update(additional_context)
    if base_path is None:
        base_path = Path.cwd()
    return super().dryrun(base_path=base_path, context=combined_context, override_path=override_path)

validate(base_path=None, override_path=None, **additional_context)

Validate the created repo.

Source code in src/nskit/mixer/components/recipe.py 
227
228
229
230
231
232
233
def validate(self, base_path: Optional[Path] = None, override_path: Optional[Path] = None, **additional_context):
    """Validate the created repo."""
    combined_context = self.context
    combined_context.update(additional_context)
    if base_path is None:
        base_path = Path.cwd()
    return super().validate(base_path=base_path, context=combined_context, override_path=override_path)

load(recipe_name, entrypoint=None, initialize=True, **kwargs) staticmethod

Load a recipe as an extension.

Parameters:

Name Type Description Default
recipe_name str

Name of the recipe to load

 required
entrypoint Optional[str]

Recipe entrypoint to use (defaults to RECIPE_ENTRYPOINT)

 None
initialize bool

Whether to initialize the recipe instance (default True)

 True
**kwargs

Arguments to pass to recipe initialization

 {}

Returns:

Type Description

Recipe instance if initialize=True, otherwise recipe class
Source code in src/nskit/mixer/components/recipe.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
@staticmethod
def load(recipe_name: str, entrypoint: Optional[str] = None, initialize: bool = True, **kwargs):
    """Load a recipe as an extension.

    Args:
        recipe_name: Name of the recipe to load
        entrypoint: Recipe entrypoint to use (defaults to RECIPE_ENTRYPOINT)
        initialize: Whether to initialize the recipe instance (default True)
        **kwargs: Arguments to pass to recipe initialization

    Returns:
        Recipe instance if initialize=True, otherwise recipe class
    """
    if entrypoint is None:
        entrypoint = RECIPE_ENTRYPOINT

    recipe_klass = load_extension(entrypoint, recipe_name)
    if recipe_klass is None:
        raise ValueError(
            f"Recipe {recipe_name} not found, it may be mis-spelt or not installed. Available recipes: {get_extension_names(entrypoint)}"
        )

    if not initialize:
        recipe_klass.extension_name = recipe_name
        return recipe_klass

    recipe = recipe_klass(**kwargs)
    recipe.extension_name = recipe_name
    return recipe

inspect(recipe_name, entrypoint=None, include_private=False, include_folder=False, include_base=False) staticmethod

Get the fields on a recipe as an extension.

Parameters:

Name Type Description Default
recipe_name str

Name of the recipe to inspect

 required
entrypoint Optional[str]

Recipe entrypoint to use (defaults to RECIPE_ENTRYPOINT)

 None
include_private bool

Include private fields

 False
include_folder bool

Include folder fields

 False
include_base bool

Include base recipe fields

 False

Returns:

Type Description

Signature of the recipe
Source code in src/nskit/mixer/components/recipe.py 
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
@staticmethod
def inspect(
    recipe_name: str,
    entrypoint: Optional[str] = None,
    include_private: bool = False,
    include_folder: bool = False,
    include_base: bool = False,
):
    """Get the fields on a recipe as an extension.

    Args:
        recipe_name: Name of the recipe to inspect
        entrypoint: Recipe entrypoint to use (defaults to RECIPE_ENTRYPOINT)
        include_private: Include private fields
        include_folder: Include folder fields
        include_base: Include base recipe fields

    Returns:
        Signature of the recipe
    """
    if entrypoint is None:
        entrypoint = RECIPE_ENTRYPOINT

    recipe_klass = load_extension(entrypoint, recipe_name)
    if recipe_klass is None:
        raise ValueError(
            f"Recipe {recipe_name} not found, it may be mis-spelt or not installed. Available recipes: {get_extension_names(entrypoint)}"
        )
    sig = Recipe._inspect_basemodel(recipe_klass, include_private=include_private)
    if not include_folder:
        folder_sig = inspect.signature(Folder)
        params = [v for u, v in sig.parameters.items() if u not in folder_sig.parameters.keys() or u == "name"]
        sig = sig.replace(parameters=params)
    if not include_base:
        recipe_sig = inspect.signature(Recipe)
        params = [v for u, v in sig.parameters.items() if u not in recipe_sig.parameters.keys() or u == "name"]
        sig = sig.replace(parameters=params)
    return sig

nskit.mixer.components.recipe.RecipeField(default=..., *, env_var=None, template=None, prompt_text=None, display_name=None, help_text=None, options=None, options_provider=None, default_provider=None, conditional_rules=None, description=None, **kwargs)

Convenience wrapper around pydantic.Field for recipe fields.

Packs interactive metadata into json_schema_extra so it is available for FieldParser.from_recipe_model() introspection without affecting Pydantic validation.

Parameters:

Name Type Description Default
default Any

Default value for the field.

 ...
env_var Optional[str]

Environment variable name for default resolution.

 None
template Optional[str]

Jinja2 template expression for derived defaults.

 None
prompt_text Optional[str]

Custom prompt text for interactive collection.

 None
display_name Optional[str]

Human-readable field name for UI prompting.

 None
help_text Optional[str]

Additional help text shown alongside the prompt.

 None
options Optional[list[str]]

Available choices for enum-type fields.

 None
options_provider Optional[str]

Named provider for dynamically resolved options. The InteractiveHandler looks up this name in its options_providers registry and calls the corresponding callable at prompt time to populate options.

 None
default_provider Optional[str]

Named provider for dynamically resolved defaults. The InteractiveHandler looks up this name in its default_providers registry and calls the corresponding callable at prompt time (after env_var and template resolution) to produce a default value.

 None
conditional_rules Optional[list[dict]]

Rules controlling field visibility. Each may use the structured {depends_on, operator, value} form or the "op:value" condition shorthand.

 None
description Optional[str]

Description of the field's purpose.

 None
**kwargs Any

Additional keyword arguments passed to pydantic.Field.

 {}

Returns:

Type Description
FieldInfo

A Pydantic FieldInfo instance with interactive metadata.
Source code in src/nskit/mixer/components/recipe.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def RecipeField(
    default: Any = ...,
    *,
    env_var: Optional[str] = None,
    template: Optional[str] = None,
    prompt_text: Optional[str] = None,
    display_name: Optional[str] = None,
    help_text: Optional[str] = None,
    options: Optional[list[str]] = None,
    options_provider: Optional[str] = None,
    default_provider: Optional[str] = None,
    conditional_rules: Optional[list[dict]] = None,
    description: Optional[str] = None,
    **kwargs: Any,
) -> FieldInfo:
    """Convenience wrapper around ``pydantic.Field`` for recipe fields.

    Packs interactive metadata into ``json_schema_extra`` so it is available
    for ``FieldParser.from_recipe_model()`` introspection without affecting
    Pydantic validation.

    Args:
        default: Default value for the field.
        env_var: Environment variable name for default resolution.
        template: Jinja2 template expression for derived defaults.
        prompt_text: Custom prompt text for interactive collection.
        display_name: Human-readable field name for UI prompting.
        help_text: Additional help text shown alongside the prompt.
        options: Available choices for enum-type fields.
        options_provider: Named provider for dynamically resolved options.
            The ``InteractiveHandler`` looks up this name in its
            ``options_providers`` registry and calls the corresponding
            callable at prompt time to populate ``options``.
        default_provider: Named provider for dynamically resolved defaults.
            The ``InteractiveHandler`` looks up this name in its
            ``default_providers`` registry and calls the corresponding
            callable at prompt time (after ``env_var`` and ``template``
            resolution) to produce a default value.
        conditional_rules: Rules controlling field visibility. Each may use the
            structured ``{depends_on, operator, value}`` form or the ``"op:value"``
            ``condition`` shorthand.
        description: Description of the field's purpose.
        **kwargs: Additional keyword arguments passed to ``pydantic.Field``.

    Returns:
        A Pydantic ``FieldInfo`` instance with interactive metadata.
    """
    extra: dict[str, Any] = {}
    if env_var is not None:
        extra["env_var"] = env_var
    if template is not None:
        extra["template"] = template
    if prompt_text is not None:
        extra["prompt_text"] = prompt_text
    if display_name is not None:
        extra["display_name"] = display_name
    if help_text is not None:
        extra["help_text"] = help_text
    if options is not None:
        extra["options"] = options
    if options_provider is not None:
        extra["options_provider"] = options_provider
    if default_provider is not None:
        extra["default_provider"] = default_provider
    if conditional_rules is not None:
        extra["conditional_rules"] = conditional_rules
    return Field(
        default,
        description=description,
        json_schema_extra=extra or None,
        **kwargs,
    )

nskit.mixer.components.hook.Hook

Bases: ABC, BaseModel

Hook component.

Hooks receive the recipe path and context, and optionally the recipe instance itself. The recipe kwarg enables pre-write hooks to mutate the recipe's contents before rendering.

Backwards-compatible: existing hooks that define call(self, recipe_path, context) without **kwargs continue to work — the recipe kwarg is only forwarded if the hook's call signature accepts it.

Source code in src/nskit/mixer/components/hook.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
class Hook(ABC, BaseModel):
    """Hook component.

    Hooks receive the recipe path and context, and optionally the recipe
    instance itself. The ``recipe`` kwarg enables pre-write hooks to mutate
    the recipe's ``contents`` before rendering.

    Backwards-compatible: existing hooks that define
    ``call(self, recipe_path, context)`` without **kwargs continue to work —
    the recipe kwarg is only forwarded if the hook's ``call`` signature accepts it.
    """

    @abstractmethod
    def call(self, recipe_path: Path, context: dict[str, Any], **kwargs) -> Optional[tuple[Path, dict]]:
        """Execute the hook logic.

        Args:
            recipe_path: Path where the recipe will be (pre) or was (post) written.
            context: Template rendering context (all recipe fields + properties).
            **kwargs: Additional keyword arguments. Currently passes ``recipe``
                (the Recipe instance) when called from ``Recipe.create()``.
                Hooks that don't need the recipe can ignore it via **kwargs.

        Returns:
            None to keep path/context unchanged, or ``(recipe_path, context)`` tuple.
        """
        raise NotImplementedError()

    def __call__(self, recipe_path: Path, context: dict[str, Any], **kwargs) -> tuple[Path, dict]:
        """Call the hook and return tuple (recipe_path, context).

        Inspects the ``call`` method signature to determine whether to forward
        kwargs (like ``recipe``). This ensures backwards compatibility with
        existing hooks that only accept ``(recipe_path, context)``.
        """
        sig = inspect.signature(self.call)
        params = sig.parameters

        # Forward kwargs only if call() accepts **kwargs or explicitly declares the kwarg names
        accepts_kwargs = any(p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values())
        if accepts_kwargs:
            hook_result = self.call(recipe_path, context, **kwargs)
        else:
            # Check which specific kwargs the method accepts
            forward = {k: v for k, v in kwargs.items() if k in params}
            hook_result = self.call(recipe_path, context, **forward)

        if hook_result:
            recipe_path, context = hook_result
        return recipe_path, context

__call__(recipe_path, context, **kwargs)

Call the hook and return tuple (recipe_path, context).

Inspects the call method signature to determine whether to forward kwargs (like recipe). This ensures backwards compatibility with existing hooks that only accept (recipe_path, context).

Source code in src/nskit/mixer/components/hook.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def __call__(self, recipe_path: Path, context: dict[str, Any], **kwargs) -> tuple[Path, dict]:
    """Call the hook and return tuple (recipe_path, context).

    Inspects the ``call`` method signature to determine whether to forward
    kwargs (like ``recipe``). This ensures backwards compatibility with
    existing hooks that only accept ``(recipe_path, context)``.
    """
    sig = inspect.signature(self.call)
    params = sig.parameters

    # Forward kwargs only if call() accepts **kwargs or explicitly declares the kwarg names
    accepts_kwargs = any(p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values())
    if accepts_kwargs:
        hook_result = self.call(recipe_path, context, **kwargs)
    else:
        # Check which specific kwargs the method accepts
        forward = {k: v for k, v in kwargs.items() if k in params}
        hook_result = self.call(recipe_path, context, **forward)

    if hook_result:
        recipe_path, context = hook_result
    return recipe_path, context

call(recipe_path, context, **kwargs) abstractmethod

Execute the hook logic.

Parameters:

Name Type Description Default
recipe_path Path

Path where the recipe will be (pre) or was (post) written.

 required
context dict[str, Any]

Template rendering context (all recipe fields + properties).

 required
**kwargs

Additional keyword arguments. Currently passes recipe (the Recipe instance) when called from Recipe.create(). Hooks that don't need the recipe can ignore it via **kwargs.

 {}

Returns:

Type Description
Optional[tuple[Path, dict]]

None to keep path/context unchanged, or (recipe_path, context) tuple.
Source code in src/nskit/mixer/components/hook.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@abstractmethod
def call(self, recipe_path: Path, context: dict[str, Any], **kwargs) -> Optional[tuple[Path, dict]]:
    """Execute the hook logic.

    Args:
        recipe_path: Path where the recipe will be (pre) or was (post) written.
        context: Template rendering context (all recipe fields + properties).
        **kwargs: Additional keyword arguments. Currently passes ``recipe``
            (the Recipe instance) when called from ``Recipe.create()``.
            Hooks that don't need the recipe can ignore it via **kwargs.

    Returns:
        None to keep path/context unchanged, or ``(recipe_path, context)`` tuple.
    """
    raise NotImplementedError()

nskit.mixer.components.license_file.LicenseFile

Bases: File

License File created by downloading from Github.

Source code in src/nskit/mixer/components/license_file.py 
127
128
129
130
131
132
133
class LicenseFile(File):
    """License File created by downloading from Github."""

    name: Optional[Union[TemplateStr, str, Callable]] = Field(
        get_license_filename, validate_default=True, description="The name of the license file"
    )
    content: Union[Resource, str, bytes, Path, Callable] = Field(get_license_content, description="The file content")

nskit.mixer.components.license_file.LicenseOptionsEnum

Bases: Enum

License Options for the license file.

Built from Github API licenses.get_all_commonly_used.

Source code in src/nskit/mixer/components/license_file.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class LicenseOptionsEnum(Enum):
    """License Options for the license file.

    Built from Github API licenses.get_all_commonly_used.
    """

    AGPL_3_0 = "agpl-3.0"
    Apache_2_0 = "apache-2.0"
    BSD_2_Clause = "bsd-2-clause"
    BSD_3_Clause = "bsd-3-clause"
    BSL_1_0 = "bsl-1.0"
    CC0_1_0 = "cc0-1.0"
    EPL_2_0 = "epl-2.0"
    GPL_2_0 = "gpl-2.0"
    GPL_3_0 = "gpl-3.0"
    LGPL_2_1 = "lgpl-2.1"
    MIT = "mit"
    MPL_2_0 = "mpl-2.0"
    Unlicense = "unlicense"

    @classmethod
    def contains(cls, value):
        """Return True if `value` is in `cls`.

        BACKPORT FROM PYTHON 3.12

        `value` is in `cls` if:
        1) `value` is a member of `cls`, or
        2) `value` is the value of one of the `cls`'s members.
        """
        if sys.version_info.major <= 3 and sys.version_info.minor < 12:
            if isinstance(value, cls):
                return True
            try:
                return value in cls._value2member_map_
            except TypeError:
                return value in cls._unhashable_values_
        return value in cls

contains(value) classmethod

Return True if value is in cls.

BACKPORT FROM PYTHON 3.12

value is in cls if: 1) value is a member of cls, or 2) value is the value of one of the cls's members.

Source code in src/nskit/mixer/components/license_file.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@classmethod
def contains(cls, value):
    """Return True if `value` is in `cls`.

    BACKPORT FROM PYTHON 3.12

    `value` is in `cls` if:
    1) `value` is a member of `cls`, or
    2) `value` is the value of one of the `cls`'s members.
    """
    if sys.version_info.major <= 3 and sys.version_info.minor < 12:
        if isinstance(value, cls):
            return True
        try:
            return value in cls._value2member_map_
        except TypeError:
            return value in cls._unhashable_values_
    return value in cls

Repo Metadata

nskit.mixer.repo.CodeRecipe

Bases: Recipe

Recipe for a (git) code repo.

Language-agnostic: includes a default git init hook only. Language-specific recipes (e.g. PyRecipe) extend post_hooks with their own tooling (such as pre-commit installation) and set language accordingly.

Source code in src/nskit/mixer/repo.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
class CodeRecipe(Recipe):
    """Recipe for a (git) code repo.

    Language-agnostic: includes a default git init hook only. Language-specific
    recipes (e.g. ``PyRecipe``) extend ``post_hooks`` with their own tooling
    (such as pre-commit installation) and set ``language`` accordingly.
    """

    repo: RepoMetadata
    post_hooks: Optional[list[Callable]] = Field(
        [hooks.git.GitInit()],
        validate_default=True,
        description="Hooks that can be used to modify a recipe path and context after writing",
    )
    git: GitConfig = GitConfig()
    language: Optional[str] = Field(
        None, description="The primary language of the repo, set by the language-specific recipe"
    )
    license: Optional[LicenseOptionsEnum] = None

    def get_pipeline_filenames(self):
        """Get CICD Pipeline filenames."""
        return []

get_pipeline_filenames()

Get CICD Pipeline filenames.

Source code in src/nskit/mixer/repo.py 
65
66
67
def get_pipeline_filenames(self):
    """Get CICD Pipeline filenames."""
    return []

nskit.mixer.repo.RepoMetadata

Bases: BaseConfiguration

Repository/package metadata information.

Source code in src/nskit/mixer/repo.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class RepoMetadata(BaseConfiguration):
    """Repository/package metadata information."""

    repo_separator: str = "-"
    owner: str = Field(..., description="Who is the owner of the repo")
    email: EmailStr = Field(..., description="The email for the repo owner")
    description: str = Field("", description="A summary description for the repo")
    url: HttpUrl = Field(..., description="The Repository url.")

    @field_validator("url", mode="before")
    @classmethod
    def _prepend_scheme(cls, v):
        if isinstance(v, str) and not v.startswith("https://"):
            v = v.removeprefix("http://")
            v = f"https://{v}"
        return v

Hooks

nskit.mixer.hooks.git.GitInit

Bases: Hook

Git Hook to (re) initialise a repo.

Source code in src/nskit/mixer/hooks/git.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class GitInit(Hook):
    """Git Hook to (re) initialise a repo."""

    def call(self, recipe_path: Path, context: dict[str, Any]):
        """(re)initialise the repo."""
        with ChDir(recipe_path):
            logger.info("Initialising git repo")
            try:
                initial_branch_name = subprocess.check_output(["git", "config", "--get", "init.defaultBranch"]).decode()  # nosec B607, B603
            except subprocess.CalledProcessError:
                initial_branch_name = None
            if not initial_branch_name:
                initial_branch_name = "main"
            initial_branch_name = context.get("git", {}).get("initial_branch_name", initial_branch_name)
            # Validate branch name to prevent argument injection
            initial_branch_name = initial_branch_name.strip()
            if not initial_branch_name or initial_branch_name.startswith("-"):
                initial_branch_name = "main"
            # Check git version - new versions have --initial-branch arg on init
            version = subprocess.check_output(["git", "version"]).decode()  # nosec B607, B603
            version = version.replace("git version", "").lstrip()
            semver = parse(".".join(version.split(" ")[0].split(".")[:3]))
            if semver >= parse("2.28.0"):
                subprocess.check_call(["git", "init", "--initial-branch", initial_branch_name])  # nosec B607, B603
            else:
                subprocess.check_call(["git", "init"])  # nosec B607, B603
                subprocess.check_call(["git", "checkout", "-B", initial_branch_name])  # nosec B607, B603
            logger.info("Done")

call(recipe_path, context)

(re)initialise the repo.

Source code in src/nskit/mixer/hooks/git.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def call(self, recipe_path: Path, context: dict[str, Any]):
    """(re)initialise the repo."""
    with ChDir(recipe_path):
        logger.info("Initialising git repo")
        try:
            initial_branch_name = subprocess.check_output(["git", "config", "--get", "init.defaultBranch"]).decode()  # nosec B607, B603
        except subprocess.CalledProcessError:
            initial_branch_name = None
        if not initial_branch_name:
            initial_branch_name = "main"
        initial_branch_name = context.get("git", {}).get("initial_branch_name", initial_branch_name)
        # Validate branch name to prevent argument injection
        initial_branch_name = initial_branch_name.strip()
        if not initial_branch_name or initial_branch_name.startswith("-"):
            initial_branch_name = "main"
        # Check git version - new versions have --initial-branch arg on init
        version = subprocess.check_output(["git", "version"]).decode()  # nosec B607, B603
        version = version.replace("git version", "").lstrip()
        semver = parse(".".join(version.split(" ")[0].split(".")[:3]))
        if semver >= parse("2.28.0"):
            subprocess.check_call(["git", "init", "--initial-branch", initial_branch_name])  # nosec B607, B603
        else:
            subprocess.check_call(["git", "init"])  # nosec B607, B603
            subprocess.check_call(["git", "checkout", "-B", initial_branch_name])  # nosec B607, B603
        logger.info("Done")

nskit.mixer.hooks.pre_commit.PrecommitInstall

Bases: Hook

Precommit install hook.

Source code in src/nskit/mixer/hooks/pre_commit.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
class PrecommitInstall(Hook):
    """Precommit install hook."""

    def call(self, recipe_path: Path, context: dict[str, Any]):  # noqa: U100
        """Run the pre-commit install and install hooks command."""
        with ChDir(recipe_path):
            if Path(".pre-commit-config.yaml").exists():
                logger.info("Installing precommit")
                # Try uv first, fall back to pip
                try:
                    subprocess.check_call(
                        [sys.executable, "-m", "pip", "install", "pre-commit"],  # nosec B603
                        stdout=subprocess.DEVNULL,
                        stderr=subprocess.DEVNULL,
                    )
                except (subprocess.CalledProcessError, FileNotFoundError):
                    try:
                        subprocess.check_call(
                            ["uv", "pip", "install", "pre-commit"],  # nosec B603, B607
                            stdout=subprocess.DEVNULL,
                            stderr=subprocess.DEVNULL,
                        )
                    except (subprocess.CalledProcessError, FileNotFoundError):
                        logger.warning("Could not install pre-commit, skipping hook installation.")
                        return
                logger.info("Installing hooks")
                with open(".pre-commit-config.yaml") as f:
                    logger.info(f"Precommit Config: {f.read()}")
                # Run
                try:
                    subprocess.check_output([sys.executable, "-m", "pre_commit", "install", "--install-hooks"])  # nosec B603
                except subprocess.CalledProcessError as e:
                    logger.error("Error running pre-commit", output=e.output, return_code=e.returncode)
                    raise e from None
                logger.info("Done")
            else:
                logger.info("Precommit config file not detected, skipping.")

call(recipe_path, context)

Run the pre-commit install and install hooks command.

Source code in src/nskit/mixer/hooks/pre_commit.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def call(self, recipe_path: Path, context: dict[str, Any]):  # noqa: U100
    """Run the pre-commit install and install hooks command."""
    with ChDir(recipe_path):
        if Path(".pre-commit-config.yaml").exists():
            logger.info("Installing precommit")
            # Try uv first, fall back to pip
            try:
                subprocess.check_call(
                    [sys.executable, "-m", "pip", "install", "pre-commit"],  # nosec B603
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.DEVNULL,
                )
            except (subprocess.CalledProcessError, FileNotFoundError):
                try:
                    subprocess.check_call(
                        ["uv", "pip", "install", "pre-commit"],  # nosec B603, B607
                        stdout=subprocess.DEVNULL,
                        stderr=subprocess.DEVNULL,
                    )
                except (subprocess.CalledProcessError, FileNotFoundError):
                    logger.warning("Could not install pre-commit, skipping hook installation.")
                    return
            logger.info("Installing hooks")
            with open(".pre-commit-config.yaml") as f:
                logger.info(f"Precommit Config: {f.read()}")
            # Run
            try:
                subprocess.check_output([sys.executable, "-m", "pre_commit", "install", "--install-hooks"])  # nosec B603
            except subprocess.CalledProcessError as e:
                logger.error("Error running pre-commit", output=e.output, return_code=e.returncode)
                raise e from None
            logger.info("Done")
        else:
            logger.info("Precommit config file not detected, skipping.")

Utilities

nskit.mixer.utilities.Resource

Bases: str

A type for a package resource uri.

Source code in src/nskit/mixer/utilities.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
class Resource(str):
    """A type for a package resource uri."""

    @classmethod
    def __get_pydantic_core_schema__(
        cls,
        source_type: Any,
        handler: GetCoreSchemaHandler,  # noqa: U100
    ) -> CoreSchema:
        """Get the schema."""
        return core_schema.no_info_after_validator_function(cls._validate_resource, handler(str))

    @classmethod
    def _validate_resource(cls, value: str):
        resource_string_example = "<package>.<module>:<resource filename>"
        if ":" not in value:
            raise ValueError(f"Value should be a resource string, looking like {resource_string_example}.")
        parts = value.split(":")
        if len(parts) != 2:
            raise ValueError(f"Value should be a resource string, looking like {resource_string_example}.")
        path, filename = parts
        # filename should be a valid filename
        if len(Path(filename).parts) != 1:
            raise ValueError(
                f"The part after the colon ({filename}) should be a valid filename as part of the resource string ({resource_string_example})."
            )
        # path should be a valid module path
        invalid_path = [u in path for u in [" ", "-", "*", "(", ")"]]
        if any(invalid_path):
            raise ValueError(
                f"The part before the colon ({path}) should be a valid python module path as part of the resource string ({resource_string_example})."
            )
        return cls(value)

    def load(self):
        """Load the resource using importlib.resources."""
        path, filename = self.split(":")
        p = files(path).joinpath(filename)
        return p.read_text(encoding="utf-8")

    @classmethod
    def validate(cls, value):
        """Validate the input."""
        ta = TypeAdapter(Resource)
        return ta.validate_python(value)

__get_pydantic_core_schema__(source_type, handler) classmethod

Get the schema.

Source code in src/nskit/mixer/utilities.py 
24
25
26
27
28
29
30
31
@classmethod
def __get_pydantic_core_schema__(
    cls,
    source_type: Any,
    handler: GetCoreSchemaHandler,  # noqa: U100
) -> CoreSchema:
    """Get the schema."""
    return core_schema.no_info_after_validator_function(cls._validate_resource, handler(str))

load()

Load the resource using importlib.resources.

Source code in src/nskit/mixer/utilities.py 
55
56
57
58
59
def load(self):
    """Load the resource using importlib.resources."""
    path, filename = self.split(":")
    p = files(path).joinpath(filename)
    return p.read_text(encoding="utf-8")

validate(value) classmethod

Validate the input.

Source code in src/nskit/mixer/utilities.py 
61
62
63
64
65
@classmethod
def validate(cls, value):
    """Validate the input."""
    ta = TypeAdapter(Resource)
    return ta.validate_python(value)