Skip to content

nskit.mixer

nskit.mixer.

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

nskit.mixer.components

nskit.mixer components for building recipes.

File

Bases: FileSystemObject

File component.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/file.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
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
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 .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/file.py 
49
50
51
52
53
54
55
56
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 .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/file.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
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 .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/file.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
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 .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/file.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
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

Folder

Bases: FileSystemObject

Folder component.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/folder.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
 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
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 .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/folder.py 
77
78
79
80
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 .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/folder.py 
82
83
84
85
86
87
88
89
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 .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/folder.py 
25
26
27
28
29
30
31
32
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 .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/folder.py 
70
71
72
73
74
75
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 .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/folder.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
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 .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/folder.py 
16
17
18
19
20
21
22
23
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}

Hook

Bases: ABC, BaseModel

Hook component.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/hook.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
class Hook(ABC, BaseModel):
    """Hook component."""

    @abstractmethod
    def call(self, recipe_path: Path, context: Dict[str, Any]) -> Optional[Tuple[str, Path, Dict]]:
        """Return None or tuple (recipe_path, context)."""
        raise NotImplementedError()

    def __call__(self, recipe_path: Path, context: Dict[str, Any]) -> Tuple[str, Path, Dict]:
        """Call the hook and return tuple (recipe_path, context)."""
        hook_result = self.call(recipe_path, context)
        if hook_result:
            recipe_path, context = hook_result
        return recipe_path, context

__call__(recipe_path, context)

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

Source code in .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/hook.py 
17
18
19
20
21
22
def __call__(self, recipe_path: Path, context: Dict[str, Any]) -> Tuple[str, Path, Dict]:
    """Call the hook and return tuple (recipe_path, context)."""
    hook_result = self.call(recipe_path, context)
    if hook_result:
        recipe_path, context = hook_result
    return recipe_path, context

call(recipe_path, context) abstractmethod

Return None or tuple (recipe_path, context).

Source code in .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/hook.py 
12
13
14
15
@abstractmethod
def call(self, recipe_path: Path, context: Dict[str, Any]) -> Optional[Tuple[str, Path, Dict]]:
    """Return None or tuple (recipe_path, context)."""
    raise NotImplementedError()

LicenseFile

Bases: File

License File created by downloading from Github.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/license_file.py 
119
120
121
122
123
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')

LicenseOptionsEnum

Bases: Enum

License Options for the license file.

Built from Github API licenses.get_all_commonly_used.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/license_file.py 
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
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 .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/license_file.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@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

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 .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/recipe.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
 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
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
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.")

    @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)
        content = self.write(recipe_path.parent, context, override_path=recipe_path.name)
        recipe_path = list(content.keys())[0]
        for hook in self.post_hooks:
            recipe_path, context = hook(recipe_path, context)
        self._write_batch(Path(recipe_path))
        return {Path(recipe_path): list(content.values())[0]}

    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()
        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, **kwargs):
        """Load a recipe as an extension."""
        recipe_klass = load_extension(RECIPE_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(RECIPE_ENTRYPOINT)}')
        recipe = recipe_klass(**kwargs)
        recipe.extension_name = recipe_name
        return recipe

    @staticmethod
    def inspect(recipe_name: str, include_private: bool = False, include_folder: bool = False, include_base: bool = False):
        """Get the fields on a recipe as an extension."""
        recipe_klass = load_extension(RECIPE_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(RECIPE_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)

context property

Get the context on the initialised recipe.

recipe property

Recipe context.

recipe_batch property

Get information about the specific info of this recipe.

__repr__()

Repr(x) == x.repr.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/recipe.py 
128
129
130
131
def __repr__(self):
    """Repr(x) == x.__repr__."""
    context = self.context
    return f'{self._repr(context=context)}\n\nContext: {context}'

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 .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/recipe.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
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)
    content = self.write(recipe_path.parent, context, override_path=recipe_path.name)
    recipe_path = list(content.keys())[0]
    for hook in self.post_hooks:
        recipe_path, context = hook(recipe_path, context)
    self._write_batch(Path(recipe_path))
    return {Path(recipe_path): list(content.values())[0]}

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

See the recipe as a dry run.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/recipe.py 
133
134
135
136
137
138
139
140
141
142
143
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)

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

Get the fields on a recipe as an extension.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/recipe.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
@staticmethod
def inspect(recipe_name: str, include_private: bool = False, include_folder: bool = False, include_base: bool = False):
    """Get the fields on a recipe as an extension."""
    recipe_klass = load_extension(RECIPE_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(RECIPE_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

load(recipe_name, **kwargs) staticmethod

Load a recipe as an extension.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/recipe.py 
157
158
159
160
161
162
163
164
165
@staticmethod
def load(recipe_name: str, **kwargs):
    """Load a recipe as an extension."""
    recipe_klass = load_extension(RECIPE_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(RECIPE_ENTRYPOINT)}')
    recipe = recipe_klass(**kwargs)
    recipe.extension_name = recipe_name
    return recipe

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

Validate the created repo.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/recipe.py 
145
146
147
148
149
150
151
152
153
154
155
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)

nskit.mixer.components.license_file

License file handler.

get_license_filename(context=None)

Callable to set the default license file name.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/license_file.py 
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
def get_license_filename(context: Optional[Dict[str, Any]] = None):
    """Callable to set the default license file name."""
    # Can't do in LicenseOptionsEnum pre 3.12
    if LicenseOptionsEnum.contains(context.get('license', None)):
        # Handle naming
        license_name = LicenseOptionsEnum(context.get('license', None))
        # COPYING
        if license_name in [
            LicenseOptionsEnum.AGPL_3_0,
            LicenseOptionsEnum.GPL_2_0,
            LicenseOptionsEnum.GPL_3_0,
        ]:
            name = 'COPYING'
        # COPYING.LESSER
        elif license_name in [
            LicenseOptionsEnum.LGPL_2_1,
        ]:
            name = 'COPYING.LESSER'
        # LICENSE
        elif license_name in [
            LicenseOptionsEnum.MIT,
            LicenseOptionsEnum.Apache_2_0,
            LicenseOptionsEnum.BSD_2_Clause,
            LicenseOptionsEnum.BSD_3_Clause,
            LicenseOptionsEnum.BSL_1_0,
            LicenseOptionsEnum.CC0_1_0,
            LicenseOptionsEnum.EPL_2_0,
            LicenseOptionsEnum.MIT,
            LicenseOptionsEnum.MPL_2_0
        ]:
            name = "LICENSE"
        # UNLICENSE
        elif license_name in [LicenseOptionsEnum.Unlicense]:
            name = 'UNLICENSE'
        return name

get_license_content(context)

Render the content of the license.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/mixer/components/license_file.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def get_license_content(context: Dict[str, Any]):
    """Render the content of the license."""
    # We implement some specifics based on the implementation instructions in Github licenses api get
    # Can't do in LicenseOptionsEnum pre 3.12
    if LicenseOptionsEnum.contains(context.get('license', None)):
        license_name = LicenseOptionsEnum(context.get('license', None))
        license_content = _get_license_content(license_name)
        content = license_content.body
        # [year] [fullname] to be replaced
        if license_name in [
            LicenseOptionsEnum.BSD_2_Clause,
            LicenseOptionsEnum.BSD_3_Clause,
            LicenseOptionsEnum.MIT,
        ]:
            content = content.replace('[year]', '{{license_year}}').replace('[fullname]', '{{repo.name}} Developers')
        context['license_year'] = context.get('license_year', date.today().year)
        return content

nskit.mixer.repo

A base recipe object for a code (git) repo.

CodeRecipe

Bases: Recipe

Recipe for a code repo.

Includes default git init and precommit install hooks.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/mixer/repo.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class CodeRecipe(Recipe):
    """Recipe for a code repo.

    Includes default git init and precommit install hooks.
    """
    repo: RepoMetadata
    post_hooks: Optional[List[Callable]] = Field(
        [hooks.git.GitInit(), hooks.pre_commit.PrecommitInstall()],
        validate_default=True,
        description='Hooks that can be used to modify a recipe path and context after writing'
    )
    git: GitConfig = GitConfig()
    language: str = 'python'
    license: Optional[LicenseOptionsEnum] = None

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

get_pipeline_filenames()

Get CICD Pipeline filenames.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/mixer/repo.py 
51
52
53
def get_pipeline_filenames(self):
    """Get CICD Pipeline filenames."""
    return []

RepoMetadata

Bases: BaseConfiguration

Repository/package metadata information.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/mixer/repo.py 
26
27
28
29
30
31
32
33
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.')

nskit.mixer.hooks

Common hooks.

nskit.mixer.hooks.git

Git hooks.

GitInit

Bases: Hook

Git Hook to (re) initialise a repo.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/mixer/hooks/git.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
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)
            # 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 .venv-docs/lib/python3.12/site-packages/nskit/mixer/hooks/git.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
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)
        # 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

Precommit hooks.

Contains post creation precommit install hooks.

PrecommitInstall

Bases: Hook

Precommit install hook.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/mixer/hooks/pre_commit.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
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')
                # Install if pre-commit installed
                subprocess.check_call([sys.executable, '-m', 'pip', 'install', 'pre-commit'])  # nosec B603
                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 .venv-docs/lib/python3.12/site-packages/nskit/mixer/hooks/pre_commit.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
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')
            # Install if pre-commit installed
            subprocess.check_call([sys.executable, '-m', 'pip', 'install', 'pre-commit'])  # nosec B603
            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.')