Skip to content

Recipes module initialization (moved to nskit.client).

DiscoveryClient

Client for discovering available recipes.

Source code in src/nskit/client/discovery.py 
 9
10
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
class DiscoveryClient:
    """Client for discovering available recipes."""

    def __init__(self, backend: RecipeBackend):
        """Initialize discovery client.

        Args:
            backend: Backend for recipe discovery
        """
        self.backend = backend

    def discover_recipes(
        self,
        search_term: Optional[str] = None,
    ) -> list[RecipeInfo]:
        """Discover available recipes.

        Args:
            search_term: Optional search term to filter recipes

        Returns:
            List of discovered recipes
        """
        recipes = self.backend.list_recipes()

        if search_term:
            search_lower = search_term.lower()
            recipes = [
                r
                for r in recipes
                if search_lower in r.name.lower() or (r.description and search_lower in r.description.lower())
            ]

        return recipes

    def get_recipe_info(self, recipe_name: str) -> Optional[RecipeInfo]:
        """Get detailed info for a specific recipe.

        Args:
            recipe_name: Recipe name

        Returns:
            Recipe info or None if not found
        """
        recipes = self.backend.list_recipes()

        for recipe in recipes:
            if recipe.name == recipe_name:
                return recipe

        return None

    def get_recipe_versions(self, recipe_name: str) -> list[str]:
        """Get available versions for a recipe.

        Args:
            recipe_name: Recipe name

        Returns:
            List of available versions
        """
        return self.backend.get_recipe_versions(recipe_name)

__init__(backend)

Initialize discovery client.

Parameters:

Name Type Description Default
backend RecipeBackend

Backend for recipe discovery

 required
Source code in src/nskit/client/discovery.py 
12
13
14
15
16
17
18
def __init__(self, backend: RecipeBackend):
    """Initialize discovery client.

    Args:
        backend: Backend for recipe discovery
    """
    self.backend = backend

discover_recipes(search_term=None)

Discover available recipes.

Parameters:

Name Type Description Default
search_term Optional[str]

Optional search term to filter recipes

 None

Returns:

Type Description
list[RecipeInfo]

List of discovered recipes
Source code in src/nskit/client/discovery.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def discover_recipes(
    self,
    search_term: Optional[str] = None,
) -> list[RecipeInfo]:
    """Discover available recipes.

    Args:
        search_term: Optional search term to filter recipes

    Returns:
        List of discovered recipes
    """
    recipes = self.backend.list_recipes()

    if search_term:
        search_lower = search_term.lower()
        recipes = [
            r
            for r in recipes
            if search_lower in r.name.lower() or (r.description and search_lower in r.description.lower())
        ]

    return recipes

get_recipe_info(recipe_name)

Get detailed info for a specific recipe.

Parameters:

Name Type Description Default
recipe_name str

Recipe name

 required

Returns:

Type Description
Optional[RecipeInfo]

Recipe info or None if not found
Source code in src/nskit/client/discovery.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def get_recipe_info(self, recipe_name: str) -> Optional[RecipeInfo]:
    """Get detailed info for a specific recipe.

    Args:
        recipe_name: Recipe name

    Returns:
        Recipe info or None if not found
    """
    recipes = self.backend.list_recipes()

    for recipe in recipes:
        if recipe.name == recipe_name:
            return recipe

    return None

get_recipe_versions(recipe_name)

Get available versions for a recipe.

Parameters:

Name Type Description Default
recipe_name str

Recipe name

 required

Returns:

Type Description
list[str]

List of available versions
Source code in src/nskit/client/discovery.py 
61
62
63
64
65
66
67
68
69
70
def get_recipe_versions(self, recipe_name: str) -> list[str]:
    """Get available versions for a recipe.

    Args:
        recipe_name: Recipe name

    Returns:
        List of available versions
    """
    return self.backend.get_recipe_versions(recipe_name)

DockerEngine

Bases: RecipeEngine

Execute recipes in Docker containers.

Source code in src/nskit/client/engines/docker.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
class DockerEngine(RecipeEngine):
    """Execute recipes in Docker containers."""

    def __init__(self, skip_pull: bool = False, timeouts: EngineTimeouts | dict | None = None) -> None:
        """Initialise the Docker engine.

        Args:
            skip_pull: Skip pulling the image (useful for locally built images).
            timeouts: Timeout configuration for Docker operations.
        """
        self.skip_pull = skip_pull
        if isinstance(timeouts, dict):
            self.timeouts = EngineTimeouts(**timeouts)
        else:
            self.timeouts = timeouts or EngineTimeouts()

    def execute(
        self,
        recipe: str,
        version: str,
        parameters: dict[str, Any],
        output_dir: Path,
        image_url: str | None = None,
        entrypoint: str | None = None,
    ) -> RecipeResult:
        """Execute recipe in a Docker container.

        Writes parameters to a YAML file, mounts it into the container,
        and invokes the nskit CLI ``init`` command.

        Args:
            recipe: Recipe name.
            version: Recipe version.
            parameters: Recipe input parameters.
            output_dir: Host directory for generated output.
            image_url: Docker image URL (required).
            entrypoint: Not used for Docker engine.

        Returns:
            Recipe execution result.
        """
        if not image_url:
            raise ValueError("Docker engine requires image_url")

        validate_image_url(image_url)
        validate_recipe_name(recipe)

        errors: list[str] = []
        warnings: list[str] = []

        try:
            if not self.skip_pull:
                subprocess.run(  # nosec B603, B607
                    ["docker", "pull", image_url],
                    check=True,
                    capture_output=True,
                    timeout=self.timeouts.pull,
                )

            # Write parameters as YAML (matches CLI --input-yaml-path)
            with tempfile.NamedTemporaryFile(mode="w", suffix=".yml", delete=False) as f:
                pyyaml.safe_dump(parameters, f, default_flow_style=False)
                input_file = Path(f.name)
            # Ensure readable by non-root container user
            input_file.chmod(0o644)

            try:
                output_dir.mkdir(parents=True, exist_ok=True)

                cmd = [
                    "docker",
                    "run",
                    "--rm",
                ]
                # Run as host user so output files have correct ownership
                if hasattr(os, "getuid"):
                    cmd += ["--user", f"{os.getuid()}:{os.getgid()}"]
                cmd += [
                    "-e",
                    "HOME=/tmp",
                    "-e",
                    f"LOG_JSON={os.environ.get('LOG_JSON', 'true')}",
                    "-e",
                    f"LOGLEVEL={os.environ.get('LOGLEVEL', 'INFO')}",
                    "-v",
                    f"{output_dir.absolute()}:/app/output",
                    "-v",
                    f"{input_file.absolute()}:/app/input.yml:ro",
                    image_url,
                    "init",
                    "--recipe",
                    recipe,
                    "--input-yaml-path",
                    "/app/input.yml",
                    "--output-override-path",
                    "/app/output",
                ]

                result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=self.timeouts.run)  # nosec B603, B607

                if result.stderr:
                    warnings.append(result.stderr.strip())

                # Collect created files
                files_created = [str(p.relative_to(output_dir)) for p in output_dir.rglob("*") if p.is_file()]

                return RecipeResult(
                    success=True,
                    project_path=output_dir,
                    recipe_name=recipe,
                    recipe_version=version,
                    files_created=files_created,
                    warnings=warnings,
                )
            finally:
                input_file.unlink(missing_ok=True)

        except subprocess.CalledProcessError as e:
            detail = e.stderr.strip() if e.stderr else str(e)
            errors.append(detail)
            return RecipeResult(
                success=False,
                project_path=output_dir,
                recipe_name=recipe,
                recipe_version=version,
                errors=errors,
                warnings=warnings,
            )
        except Exception as e:
            errors.append(str(e))
            return RecipeResult(
                success=False,
                project_path=output_dir,
                recipe_name=recipe,
                recipe_version=version,
                errors=errors,
                warnings=warnings,
            )

__init__(skip_pull=False, timeouts=None)

Initialise the Docker engine.

Parameters:

Name Type Description Default
skip_pull bool

Skip pulling the image (useful for locally built images).

 False
timeouts EngineTimeouts | dict | None

Timeout configuration for Docker operations.

 None
Source code in src/nskit/client/engines/docker.py 
22
23
24
25
26
27
28
29
30
31
32
33
def __init__(self, skip_pull: bool = False, timeouts: EngineTimeouts | dict | None = None) -> None:
    """Initialise the Docker engine.

    Args:
        skip_pull: Skip pulling the image (useful for locally built images).
        timeouts: Timeout configuration for Docker operations.
    """
    self.skip_pull = skip_pull
    if isinstance(timeouts, dict):
        self.timeouts = EngineTimeouts(**timeouts)
    else:
        self.timeouts = timeouts or EngineTimeouts()

execute(recipe, version, parameters, output_dir, image_url=None, entrypoint=None)

Execute recipe in a Docker container.

Writes parameters to a YAML file, mounts it into the container, and invokes the nskit CLI init command.

Parameters:

Name Type Description Default
recipe str

Recipe name.

 required
version str

Recipe version.

 required
parameters dict[str, Any]

Recipe input parameters.

 required
output_dir Path

Host directory for generated output.

 required
image_url str | None

Docker image URL (required).

 None
entrypoint str | None

Not used for Docker engine.

 None

Returns:

Type Description
RecipeResult

Recipe execution result.
Source code in src/nskit/client/engines/docker.py 
 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
def execute(
    self,
    recipe: str,
    version: str,
    parameters: dict[str, Any],
    output_dir: Path,
    image_url: str | None = None,
    entrypoint: str | None = None,
) -> RecipeResult:
    """Execute recipe in a Docker container.

    Writes parameters to a YAML file, mounts it into the container,
    and invokes the nskit CLI ``init`` command.

    Args:
        recipe: Recipe name.
        version: Recipe version.
        parameters: Recipe input parameters.
        output_dir: Host directory for generated output.
        image_url: Docker image URL (required).
        entrypoint: Not used for Docker engine.

    Returns:
        Recipe execution result.
    """
    if not image_url:
        raise ValueError("Docker engine requires image_url")

    validate_image_url(image_url)
    validate_recipe_name(recipe)

    errors: list[str] = []
    warnings: list[str] = []

    try:
        if not self.skip_pull:
            subprocess.run(  # nosec B603, B607
                ["docker", "pull", image_url],
                check=True,
                capture_output=True,
                timeout=self.timeouts.pull,
            )

        # Write parameters as YAML (matches CLI --input-yaml-path)
        with tempfile.NamedTemporaryFile(mode="w", suffix=".yml", delete=False) as f:
            pyyaml.safe_dump(parameters, f, default_flow_style=False)
            input_file = Path(f.name)
        # Ensure readable by non-root container user
        input_file.chmod(0o644)

        try:
            output_dir.mkdir(parents=True, exist_ok=True)

            cmd = [
                "docker",
                "run",
                "--rm",
            ]
            # Run as host user so output files have correct ownership
            if hasattr(os, "getuid"):
                cmd += ["--user", f"{os.getuid()}:{os.getgid()}"]
            cmd += [
                "-e",
                "HOME=/tmp",
                "-e",
                f"LOG_JSON={os.environ.get('LOG_JSON', 'true')}",
                "-e",
                f"LOGLEVEL={os.environ.get('LOGLEVEL', 'INFO')}",
                "-v",
                f"{output_dir.absolute()}:/app/output",
                "-v",
                f"{input_file.absolute()}:/app/input.yml:ro",
                image_url,
                "init",
                "--recipe",
                recipe,
                "--input-yaml-path",
                "/app/input.yml",
                "--output-override-path",
                "/app/output",
            ]

            result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=self.timeouts.run)  # nosec B603, B607

            if result.stderr:
                warnings.append(result.stderr.strip())

            # Collect created files
            files_created = [str(p.relative_to(output_dir)) for p in output_dir.rglob("*") if p.is_file()]

            return RecipeResult(
                success=True,
                project_path=output_dir,
                recipe_name=recipe,
                recipe_version=version,
                files_created=files_created,
                warnings=warnings,
            )
        finally:
            input_file.unlink(missing_ok=True)

    except subprocess.CalledProcessError as e:
        detail = e.stderr.strip() if e.stderr else str(e)
        errors.append(detail)
        return RecipeResult(
            success=False,
            project_path=output_dir,
            recipe_name=recipe,
            recipe_version=version,
            errors=errors,
            warnings=warnings,
        )
    except Exception as e:
        errors.append(str(e))
        return RecipeResult(
            success=False,
            project_path=output_dir,
            recipe_name=recipe,
            recipe_version=version,
            errors=errors,
            warnings=warnings,
        )

ExecutionMode

Bases: str, Enum

Recipe execution mode.

Source code in src/nskit/client/execution.py 
 6
 7
 8
 9
10
class ExecutionMode(str, Enum):
    """Recipe execution mode."""

    DOCKER = "docker"  # Run recipe in Docker container (default, production)
    LOCAL = "local"  # Run recipe from locally installed package (development)

LocalEngine

Bases: RecipeEngine

Execute recipes from locally installed packages.

Source code in src/nskit/client/engines/local.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
class LocalEngine(RecipeEngine):
    """Execute recipes from locally installed packages."""

    def execute(
        self,
        recipe: str,
        version: str,
        parameters: dict[str, Any],
        output_dir: Path,
        image_url: str = None,
        entrypoint: str = None,
    ) -> RecipeResult:
        """Execute recipe from installed package.

        Args:
            recipe: Recipe name.
            version: Recipe version.
            parameters: Recipe parameters.
            output_dir: Output directory.
            image_url: Not used for Local engine.
            entrypoint: Recipe entrypoint (required).

        Returns:
            Recipe execution result.
        """
        if not entrypoint:
            raise ValueError("Local engine requires entrypoint")

        errors = []
        warnings: list[str] = []

        try:
            recipe_instance = Recipe.load(recipe, entrypoint=entrypoint, **parameters)
            result = recipe_instance.create(base_path=output_dir.parent, override_path=output_dir.name)
            files_created = list(result.keys()) if result else []

            return RecipeResult(
                success=True,
                project_path=output_dir,
                recipe_name=recipe,
                recipe_version=version,
                files_created=files_created,
                warnings=warnings,
            )

        except Exception as e:
            errors.append(str(e))
            return RecipeResult(
                success=False,
                project_path=output_dir,
                recipe_name=recipe,
                recipe_version=version,
                errors=errors,
                warnings=warnings,
            )

execute(recipe, version, parameters, output_dir, image_url=None, entrypoint=None)

Execute recipe from installed package.

Parameters:

Name Type Description Default
recipe str

Recipe name.

 required
version str

Recipe version.

 required
parameters dict[str, Any]

Recipe parameters.

 required
output_dir Path

Output directory.

 required
image_url str

Not used for Local engine.

 None
entrypoint str

Recipe entrypoint (required).

 None

Returns:

Type Description
RecipeResult

Recipe execution result.
Source code in src/nskit/client/engines/local.py 
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
def execute(
    self,
    recipe: str,
    version: str,
    parameters: dict[str, Any],
    output_dir: Path,
    image_url: str = None,
    entrypoint: str = None,
) -> RecipeResult:
    """Execute recipe from installed package.

    Args:
        recipe: Recipe name.
        version: Recipe version.
        parameters: Recipe parameters.
        output_dir: Output directory.
        image_url: Not used for Local engine.
        entrypoint: Recipe entrypoint (required).

    Returns:
        Recipe execution result.
    """
    if not entrypoint:
        raise ValueError("Local engine requires entrypoint")

    errors = []
    warnings: list[str] = []

    try:
        recipe_instance = Recipe.load(recipe, entrypoint=entrypoint, **parameters)
        result = recipe_instance.create(base_path=output_dir.parent, override_path=output_dir.name)
        files_created = list(result.keys()) if result else []

        return RecipeResult(
            success=True,
            project_path=output_dir,
            recipe_name=recipe,
            recipe_version=version,
            files_created=files_created,
            warnings=warnings,
        )

    except Exception as e:
        errors.append(str(e))
        return RecipeResult(
            success=False,
            project_path=output_dir,
            recipe_name=recipe,
            recipe_version=version,
            errors=errors,
            warnings=warnings,
        )

RecipeClient

Pure Python client for recipe operations (no CLI dependencies).

Source code in src/nskit/client/recipes.py 
 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
class RecipeClient:
    """Pure Python client for recipe operations (no CLI dependencies)."""

    def __init__(self, backend: RecipeBackend, engine: RecipeEngine | None = None):
        """Initialize the recipe client.

        Args:
            backend: Backend for recipe discovery and fetching
            engine: Execution engine (defaults to DockerEngine)
        """
        self.backend = backend
        if engine is None:
            from nskit.client.backends.docker_local import DockerLocalBackend

            engine = DockerEngine(skip_pull=isinstance(backend, DockerLocalBackend))
        self.engine = engine

    def list_recipes(self) -> list[RecipeInfo]:
        """List all available recipes from the backend.

        Returns:
            List of recipe information
        """
        return self.backend.list_recipes()

    def get_recipe_versions(self, recipe: str) -> list[str]:
        """Get available versions for a specific recipe.

        Args:
            recipe: Recipe name

        Returns:
            List of available versions
        """
        return self.backend.get_recipe_versions(recipe)

    def initialize_recipe(
        self,
        recipe: str,
        version: str,
        parameters: dict[str, Any],
        output_dir: Path,
        force: bool = False,
    ) -> RecipeResult:
        """Initialize a new project from a recipe.

        Args:
            recipe: Recipe name
            version: Recipe version
            parameters: Recipe parameters
            output_dir: Output directory for the project
            force: Allow initialization in non-empty directory

        Returns:
            Result of the initialization
        """
        errors = []

        # Check output directory
        if output_dir.exists() and any(output_dir.iterdir()) and not force:
            errors.append(f"Output directory {output_dir} is not empty. Use force=True to override.")
            return RecipeResult(
                success=False,
                project_path=output_dir,
                recipe_name=recipe,
                recipe_version=version,
                errors=errors,
            )

        try:
            output_dir.mkdir(parents=True, exist_ok=True)

            # Get image URL from backend (only if backend supports it and engine needs it)
            image_url = None
            if hasattr(self.engine, "__class__") and self.engine.__class__.__name__ == "DockerEngine":
                if hasattr(self.backend, "get_image_url"):
                    image_url = self.backend.get_image_url(recipe, version)
                    if hasattr(self.backend, "pull_image"):
                        self.backend.pull_image(image_url)
                    # Read canonical recipe name from image label
                    recipe = _read_recipe_label(image_url) or recipe

            # Execute using engine
            return self.engine.execute(
                recipe=recipe,
                version=version,
                parameters=parameters,
                output_dir=output_dir,
                image_url=image_url,
                entrypoint=self.backend.entrypoint,
            )

        except Exception as e:
            errors.append(str(e))
            return RecipeResult(
                success=False,
                project_path=output_dir,
                recipe_name=recipe,
                recipe_version=version,
                errors=errors,
            )

    def create_repository(
        self,
        repo_name: str,
        project_path: Path | None = None,
        description: str | None = None,
        private: bool = True,
    ) -> tuple[bool, str]:
        """Create a remote repository and optionally push the project.

        Auto-detects the VCS provider from environment variables.
        If *project_path* is provided and contains a git repo, the
        project is committed and pushed to the new remote.

        Args:
            repo_name: Repository name.
            project_path: Local project directory to push.
            description: Repository description.
            private: Whether the repository should be private.

        Returns:
            Tuple of (success, message).
        """
        from nskit.recipes.repository_client import RepositoryClient

        client, provider = _detect_repo_client()
        if client is None:
            return False, "No VCS provider detected. Set appropriate environment variables (e.g. GITHUB_TOKEN)."

        try:
            repo_client = RepositoryClient(vcs_client=client)
            if project_path and (project_path / ".git").is_dir():
                info = repo_client.create_and_push(
                    repo_name,
                    project_path,
                    description=description,
                    private=private,
                )
                return True, f"Created and pushed to {info.url}"
            else:
                info = repo_client.create_repository(repo_name, description=description, private=private)
                return True, f"Created repository at {info.url}"
        except Exception as e:
            return False, f"Failed to create repository: {e}"

__init__(backend, engine=None)

Initialize the recipe client.

Parameters:

Name Type Description Default
backend RecipeBackend

Backend for recipe discovery and fetching

 required
engine RecipeEngine | None

Execution engine (defaults to DockerEngine)

 None
Source code in src/nskit/client/recipes.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(self, backend: RecipeBackend, engine: RecipeEngine | None = None):
    """Initialize the recipe client.

    Args:
        backend: Backend for recipe discovery and fetching
        engine: Execution engine (defaults to DockerEngine)
    """
    self.backend = backend
    if engine is None:
        from nskit.client.backends.docker_local import DockerLocalBackend

        engine = DockerEngine(skip_pull=isinstance(backend, DockerLocalBackend))
    self.engine = engine

create_repository(repo_name, project_path=None, description=None, private=True)

Create a remote repository and optionally push the project.

Auto-detects the VCS provider from environment variables. If project_path is provided and contains a git repo, the project is committed and pushed to the new remote.

Parameters:

Name Type Description Default
repo_name str

Repository name.

 required
project_path Path | None

Local project directory to push.

 None
description str | None

Repository description.

 None
private bool

Whether the repository should be private.

 True

Returns:

Type Description
tuple[bool, str]

Tuple of (success, message).
Source code in src/nskit/client/recipes.py 
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
def create_repository(
    self,
    repo_name: str,
    project_path: Path | None = None,
    description: str | None = None,
    private: bool = True,
) -> tuple[bool, str]:
    """Create a remote repository and optionally push the project.

    Auto-detects the VCS provider from environment variables.
    If *project_path* is provided and contains a git repo, the
    project is committed and pushed to the new remote.

    Args:
        repo_name: Repository name.
        project_path: Local project directory to push.
        description: Repository description.
        private: Whether the repository should be private.

    Returns:
        Tuple of (success, message).
    """
    from nskit.recipes.repository_client import RepositoryClient

    client, provider = _detect_repo_client()
    if client is None:
        return False, "No VCS provider detected. Set appropriate environment variables (e.g. GITHUB_TOKEN)."

    try:
        repo_client = RepositoryClient(vcs_client=client)
        if project_path and (project_path / ".git").is_dir():
            info = repo_client.create_and_push(
                repo_name,
                project_path,
                description=description,
                private=private,
            )
            return True, f"Created and pushed to {info.url}"
        else:
            info = repo_client.create_repository(repo_name, description=description, private=private)
            return True, f"Created repository at {info.url}"
    except Exception as e:
        return False, f"Failed to create repository: {e}"

get_recipe_versions(recipe)

Get available versions for a specific recipe.

Parameters:

Name Type Description Default
recipe str

Recipe name

 required

Returns:

Type Description
list[str]

List of available versions
Source code in src/nskit/client/recipes.py 
57
58
59
60
61
62
63
64
65
66
def get_recipe_versions(self, recipe: str) -> list[str]:
    """Get available versions for a specific recipe.

    Args:
        recipe: Recipe name

    Returns:
        List of available versions
    """
    return self.backend.get_recipe_versions(recipe)

initialize_recipe(recipe, version, parameters, output_dir, force=False)

Initialize a new project from a recipe.

Parameters:

Name Type Description Default
recipe str

Recipe name

 required
version str

Recipe version

 required
parameters dict[str, Any]

Recipe parameters

 required
output_dir Path

Output directory for the project

 required
force bool

Allow initialization in non-empty directory

 False

Returns:

Type Description
RecipeResult

Result of the initialization
Source code in src/nskit/client/recipes.py 
 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
def initialize_recipe(
    self,
    recipe: str,
    version: str,
    parameters: dict[str, Any],
    output_dir: Path,
    force: bool = False,
) -> RecipeResult:
    """Initialize a new project from a recipe.

    Args:
        recipe: Recipe name
        version: Recipe version
        parameters: Recipe parameters
        output_dir: Output directory for the project
        force: Allow initialization in non-empty directory

    Returns:
        Result of the initialization
    """
    errors = []

    # Check output directory
    if output_dir.exists() and any(output_dir.iterdir()) and not force:
        errors.append(f"Output directory {output_dir} is not empty. Use force=True to override.")
        return RecipeResult(
            success=False,
            project_path=output_dir,
            recipe_name=recipe,
            recipe_version=version,
            errors=errors,
        )

    try:
        output_dir.mkdir(parents=True, exist_ok=True)

        # Get image URL from backend (only if backend supports it and engine needs it)
        image_url = None
        if hasattr(self.engine, "__class__") and self.engine.__class__.__name__ == "DockerEngine":
            if hasattr(self.backend, "get_image_url"):
                image_url = self.backend.get_image_url(recipe, version)
                if hasattr(self.backend, "pull_image"):
                    self.backend.pull_image(image_url)
                # Read canonical recipe name from image label
                recipe = _read_recipe_label(image_url) or recipe

        # Execute using engine
        return self.engine.execute(
            recipe=recipe,
            version=version,
            parameters=parameters,
            output_dir=output_dir,
            image_url=image_url,
            entrypoint=self.backend.entrypoint,
        )

    except Exception as e:
        errors.append(str(e))
        return RecipeResult(
            success=False,
            project_path=output_dir,
            recipe_name=recipe,
            recipe_version=version,
            errors=errors,
        )

list_recipes()

List all available recipes from the backend.

Returns:

Type Description
list[RecipeInfo]

List of recipe information
Source code in src/nskit/client/recipes.py 
49
50
51
52
53
54
55
def list_recipes(self) -> list[RecipeInfo]:
    """List all available recipes from the backend.

    Returns:
        List of recipe information
    """
    return self.backend.list_recipes()

RecipeInfo

Bases: BaseModel

Information about a recipe.

Source code in src/nskit/client/models.py 
10
11
12
13
14
15
16
class RecipeInfo(BaseModel):
    """Information about a recipe."""

    name: str
    versions: list[str]
    description: Optional[str] = None
    metadata: dict[str, Any] = Field(default_factory=dict)

RecipeResult

Bases: BaseModel

Result of recipe initialization.

Source code in src/nskit/client/models.py 
19
20
21
22
23
24
25
26
27
28
class RecipeResult(BaseModel):
    """Result of recipe initialization."""

    success: bool
    project_path: Path
    recipe_name: str
    recipe_version: str
    files_created: list[Path] = Field(default_factory=list)
    errors: list[str] = Field(default_factory=list)
    warnings: list[str] = Field(default_factory=list)

RepositoryClient

Client for managing recipe repositories.

Source code in src/nskit/recipes/repository_client.py 
 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
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
class RepositoryClient:
    """Client for managing recipe repositories."""

    def __init__(self, vcs_client: RepoClient | None = None):
        """Initialise repository client.

        Args:
            vcs_client: Optional VCS client (e.g., GithubRepoClient)
        """
        self.vcs_client = vcs_client

    def create_repository(
        self,
        repo_name: str,
        description: str | None = None,
        private: bool = True,
    ) -> RepositoryInfo:
        """Create a new remote repository.

        Args:
            repo_name: Repository name.
            description: Repository description.
            private: Whether repository is private.

        Returns:
            Repository info including the remote URL.
        """
        if not self.vcs_client:
            raise ValueError("VCS client not configured")

        self.vcs_client.create(repo_name)
        url = str(self.vcs_client.get_remote_url(repo_name))

        return RepositoryInfo(
            name=repo_name,
            url=url,
            description=description,
        )

    def create_and_push(
        self,
        repo_name: str,
        project_path: Path,
        description: str | None = None,
        private: bool = True,
    ) -> RepositoryInfo:
        """Create a remote repository and push an existing local repo to it.

        Expects the project to already have a git repo with at least
        one commit.

        Args:
            repo_name: Repository name.
            project_path: Local project directory to push.
            description: Repository description.
            private: Whether repository is private.

        Returns:
            Repository info including the remote URL.
        """
        info = self.create_repository(repo_name, description=description, private=private)
        clone_url = str(self.vcs_client.get_clone_url(repo_name))

        subprocess.run(  # nosec B603, B607
            ["git", "remote", "add", "origin", clone_url],
            cwd=project_path,
            capture_output=True,
            check=True,
        )
        subprocess.run(  # nosec B603, B607
            ["git", "push", "-u", "origin", "HEAD"],
            cwd=project_path,
            capture_output=True,
            check=True,
        )

        return info

    def configure_repository(
        self,
        repo_name: str,
        settings: dict[str, Any] | None = None,
        default_branch: str | None = None,
        branch_rules: dict[str, Any] | None = None,
    ) -> None:
        """Apply repository configuration and branch protection via the provider.

        Delegates to the VCS provider's ``configure`` and
        ``set_branch_protection`` hooks. Providers that do not support remote
        configuration treat these as no-ops, so this is safe to call always.

        Args:
            repo_name: Repository name.
            settings: Repository-level settings overrides (merge options, features).
            default_branch: Branch to apply protection to. If ``None``, branch
                protection is skipped.
            branch_rules: Branch protection overrides for ``default_branch``.
        """
        if not self.vcs_client:
            raise ValueError("VCS client not configured")

        self.vcs_client.configure(repo_name, settings=settings)
        if default_branch is not None:
            self.vcs_client.set_branch_protection(repo_name, default_branch, rules=branch_rules)

    def get_repository_info(self, repo_name: str) -> RepositoryInfo | None:
        """Get repository information.

        Args:
            repo_name: Repository name.

        Returns:
            Repository info or ``None`` if not found.
        """
        if not self.vcs_client:
            return None

        url = str(self.vcs_client.get_remote_url(repo_name))
        return RepositoryInfo(name=repo_name, url=url)

__init__(vcs_client=None)

Initialise repository client.

Parameters:

Name Type Description Default
vcs_client RepoClient | None

Optional VCS client (e.g., GithubRepoClient)

 None
Source code in src/nskit/recipes/repository_client.py 
16
17
18
19
20
21
22
def __init__(self, vcs_client: RepoClient | None = None):
    """Initialise repository client.

    Args:
        vcs_client: Optional VCS client (e.g., GithubRepoClient)
    """
    self.vcs_client = vcs_client

configure_repository(repo_name, settings=None, default_branch=None, branch_rules=None)

Apply repository configuration and branch protection via the provider.

Delegates to the VCS provider's configure and set_branch_protection hooks. Providers that do not support remote configuration treat these as no-ops, so this is safe to call always.

Parameters:

Name Type Description Default
repo_name str

Repository name.

 required
settings dict[str, Any] | None

Repository-level settings overrides (merge options, features).

 None
default_branch str | None

Branch to apply protection to. If None, branch protection is skipped.

 None
branch_rules dict[str, Any] | None

Branch protection overrides for default_branch.

 None
Source code in src/nskit/recipes/repository_client.py 
 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
def configure_repository(
    self,
    repo_name: str,
    settings: dict[str, Any] | None = None,
    default_branch: str | None = None,
    branch_rules: dict[str, Any] | None = None,
) -> None:
    """Apply repository configuration and branch protection via the provider.

    Delegates to the VCS provider's ``configure`` and
    ``set_branch_protection`` hooks. Providers that do not support remote
    configuration treat these as no-ops, so this is safe to call always.

    Args:
        repo_name: Repository name.
        settings: Repository-level settings overrides (merge options, features).
        default_branch: Branch to apply protection to. If ``None``, branch
            protection is skipped.
        branch_rules: Branch protection overrides for ``default_branch``.
    """
    if not self.vcs_client:
        raise ValueError("VCS client not configured")

    self.vcs_client.configure(repo_name, settings=settings)
    if default_branch is not None:
        self.vcs_client.set_branch_protection(repo_name, default_branch, rules=branch_rules)

create_and_push(repo_name, project_path, description=None, private=True)

Create a remote repository and push an existing local repo to it.

Expects the project to already have a git repo with at least one commit.

Parameters:

Name Type Description Default
repo_name str

Repository name.

 required
project_path Path

Local project directory to push.

 required
description str | None

Repository description.

 None
private bool

Whether repository is private.

 True

Returns:

Type Description
RepositoryInfo

Repository info including the remote URL.
Source code in src/nskit/recipes/repository_client.py 
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
def create_and_push(
    self,
    repo_name: str,
    project_path: Path,
    description: str | None = None,
    private: bool = True,
) -> RepositoryInfo:
    """Create a remote repository and push an existing local repo to it.

    Expects the project to already have a git repo with at least
    one commit.

    Args:
        repo_name: Repository name.
        project_path: Local project directory to push.
        description: Repository description.
        private: Whether repository is private.

    Returns:
        Repository info including the remote URL.
    """
    info = self.create_repository(repo_name, description=description, private=private)
    clone_url = str(self.vcs_client.get_clone_url(repo_name))

    subprocess.run(  # nosec B603, B607
        ["git", "remote", "add", "origin", clone_url],
        cwd=project_path,
        capture_output=True,
        check=True,
    )
    subprocess.run(  # nosec B603, B607
        ["git", "push", "-u", "origin", "HEAD"],
        cwd=project_path,
        capture_output=True,
        check=True,
    )

    return info

create_repository(repo_name, description=None, private=True)

Create a new remote repository.

Parameters:

Name Type Description Default
repo_name str

Repository name.

 required
description str | None

Repository description.

 None
private bool

Whether repository is private.

 True

Returns:

Type Description
RepositoryInfo

Repository info including the remote URL.
Source code in src/nskit/recipes/repository_client.py 
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
def create_repository(
    self,
    repo_name: str,
    description: str | None = None,
    private: bool = True,
) -> RepositoryInfo:
    """Create a new remote repository.

    Args:
        repo_name: Repository name.
        description: Repository description.
        private: Whether repository is private.

    Returns:
        Repository info including the remote URL.
    """
    if not self.vcs_client:
        raise ValueError("VCS client not configured")

    self.vcs_client.create(repo_name)
    url = str(self.vcs_client.get_remote_url(repo_name))

    return RepositoryInfo(
        name=repo_name,
        url=url,
        description=description,
    )

get_repository_info(repo_name)

Get repository information.

Parameters:

Name Type Description Default
repo_name str

Repository name.

 required

Returns:

Type Description
RepositoryInfo | None

Repository info or None if not found.
Source code in src/nskit/recipes/repository_client.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def get_repository_info(self, repo_name: str) -> RepositoryInfo | None:
    """Get repository information.

    Args:
        repo_name: Repository name.

    Returns:
        Repository info or ``None`` if not found.
    """
    if not self.vcs_client:
        return None

    url = str(self.vcs_client.get_remote_url(repo_name))
    return RepositoryInfo(name=repo_name, url=url)

RepositoryInfo

Bases: BaseModel

Information about a created repository.

Source code in src/nskit/client/models.py 
44
45
46
47
48
49
50
51
52
class RepositoryInfo(BaseModel):
    """Information about a created repository."""

    name: str
    url: str
    clone_url: Optional[str] = None
    created_at: Optional[datetime] = None
    description: Optional[str] = None
    settings: dict[str, Any] = Field(default_factory=dict)

UpdateClient

Pure Python client for recipe updates (no CLI dependencies).

Parameters:

Name Type Description Default
backend RecipeBackend

Backend for fetching recipe versions.

 required
engine RecipeEngine | None

Optional recipe engine for project generation.

 None
config_dir str

Config directory name (default .recipe).

 '.recipe'
config_filename str

Config file name (default config.yml).

 'config.yml'
Source code in src/nskit/client/update.py 
 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
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
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
class UpdateClient:
    """Pure Python client for recipe updates (no CLI dependencies).

    Args:
        backend: Backend for fetching recipe versions.
        engine: Optional recipe engine for project generation.
        config_dir: Config directory name (default ``.recipe``).
        config_filename: Config file name (default ``config.yml``).
    """

    def __init__(
        self,
        backend: RecipeBackend,
        engine: RecipeEngine | None = None,
        config_dir: str = ".recipe",
        config_filename: str = "config.yml",
    ) -> None:
        self.backend = backend
        self.engine = engine
        self.config_dir = config_dir
        self.config_filename = config_filename

    def check_update_available(self, project_path: Path) -> str | None:
        """Check if an update is available for the project.

        Args:
            project_path: Path to the project.

        Returns:
            Latest version string if an update is available, ``None``
            otherwise.
        """
        config_mgr = ConfigManager(project_path, self.config_dir, self.config_filename)
        if not config_mgr.is_recipe_based():
            return None

        try:
            config = config_mgr.load_config()
            if config.metadata is None:
                return None

            resolver = VersionResolver(self.backend)
            current_image = config.metadata.docker_image
            # Extract current version from image tag
            current_version = current_image.rsplit(":", 1)[-1] if ":" in current_image else "latest"
            update_needed, resolved = resolver.check_update_needed(config.metadata.recipe_name, current_version)
            if update_needed:
                return resolved
        except Exception:  # nosec B110
            logger.warning("Could not check for updates — run with debug logging for details")
            logger.debug("Update check failed for %s", project_path, exc_info=True)

        return None

    def update_project(
        self,
        project_path: Path,
        target_version: str,
        diff_mode: DiffMode = DiffMode.THREE_WAY,
        dry_run: bool = False,
    ) -> UpdateResult:
        """Update project to target version.

        Args:
            project_path: Path to the project.
            target_version: Target version to update to.
            diff_mode: Diff mode (2-way or 3-way).
            dry_run: If ``True``, analyse changes without writing files.

        Returns:
            Update result with conflicts and errors.
        """
        # Validate git status
        git_utils = GitUtils(project_path)
        if not git_utils.is_git_repository():
            raise GitStatusError("Project is not a git repository")

        if git_utils.has_uncommitted_changes():
            raise GitStatusError("Project has uncommitted changes. Commit or stash them first.")

        config_mgr = ConfigManager(project_path, self.config_dir, self.config_filename)
        config = config_mgr.load_config()

        if config.metadata is None:
            return UpdateResult(
                success=False,
                errors=["Recipe configuration missing metadata"],
            )

        if self.engine is None:
            return UpdateResult(
                success=False,
                errors=["No recipe engine configured for project generation"],
            )

        generator = ProjectGenerator(self.backend, self.engine)
        file_discovery = FileDiscovery()
        diff_engine = DiffEngine(file_discovery=file_discovery)

        old_fresh = None
        new_fresh = None
        try:
            _, old_fresh, new_fresh = generator.generate_project_states(config, target_version, project_path, diff_mode)

            merge_result = self._extract_and_process_changes(
                project_path=project_path,
                old_fresh=old_fresh,
                new_fresh=new_fresh,
                diff_engine=diff_engine,
                git_utils=git_utils,
                diff_mode=diff_mode,
                dry_run=dry_run,
            )

            result = UpdateResult(
                success=len(merge_result.errors) == 0,
                files_updated=merge_result.clean_merges,
                files_added=merge_result.added,
                files_removed=merge_result.removed,
                files_with_conflicts=merge_result.conflicts,
                clean_merges=merge_result.clean_merges,
                errors=merge_result.errors,
            )

            if result.success and not dry_run:
                config_mgr.update_config_version(target_version, config.metadata.recipe_name)

            return result

        except UpdateError:
            raise
        except Exception as exc:
            return UpdateResult(success=False, errors=[f"Update failed: {exc}"])
        finally:
            paths_to_clean = [p for p in [old_fresh, new_fresh] if p is not None]
            if paths_to_clean:
                generator.cleanup_states(*paths_to_clean)

    def _extract_and_process_changes(
        self,
        project_path: Path,
        old_fresh: Path | None,
        new_fresh: Path,
        diff_engine: DiffEngine,
        git_utils: GitUtils,
        diff_mode: DiffMode,
        dry_run: bool,
    ) -> MergeResult:
        """Extract diffs and process file-level merges.

        Args:
            project_path: Current project path.
            old_fresh: Base version path (``None`` for 2-way).
            new_fresh: Target version path.
            diff_engine: Diff engine instance.
            git_utils: Git utilities instance.
            diff_mode: Comparison mode.
            dry_run: Whether to skip file writes.

        Returns:
            Merge result with clean merges, conflicts, and errors.
        """
        clean_merges: list[str] = []
        added: list[str] = []
        removed: list[str] = []
        conflicts: list[str] = []
        errors: list[str] = []

        if diff_mode == DiffMode.THREE_WAY and old_fresh is not None:
            diff_result = diff_engine.extract_diff(old_fresh, new_fresh, diff_mode)

            for file_diff in diff_result.modified_files:
                rel = str(file_diff.relative_path)
                result = self._process_single_file(
                    project_path,
                    old_fresh,
                    new_fresh,
                    rel,
                    git_utils,
                    dry_run,
                    three_way=True,
                )
                if result == "clean":
                    clean_merges.append(rel)
                elif result == "conflict":
                    conflicts.append(rel)
                else:
                    errors.append(result)

            for file_diff in diff_result.added_files:
                rel = str(file_diff.relative_path)
                target_file = new_fresh / rel
                dest_file = project_path / rel
                if not dry_run:
                    dest_file.parent.mkdir(parents=True, exist_ok=True)
                    dest_file.write_bytes(target_file.read_bytes())
                added.append(rel)

            for file_diff in diff_result.deleted_files:
                rel = str(file_diff.relative_path)
                dest_file = project_path / rel
                if dest_file.exists() and not dry_run:
                    dest_file.unlink()
                removed.append(rel)

        else:
            # 2-way: compare current project against target
            diff_result = diff_engine.extract_diff(project_path, new_fresh, DiffMode.TWO_WAY)

            for file_diff in diff_result.modified_files:
                rel = str(file_diff.relative_path)
                source = new_fresh / rel
                dest = project_path / rel
                if not dry_run:
                    dest.parent.mkdir(parents=True, exist_ok=True)
                    dest.write_bytes(source.read_bytes())
                clean_merges.append(rel)

            for file_diff in diff_result.added_files:
                rel = str(file_diff.relative_path)
                source = new_fresh / rel
                dest = project_path / rel
                if not dry_run:
                    dest.parent.mkdir(parents=True, exist_ok=True)
                    dest.write_bytes(source.read_bytes())
                added.append(rel)

        return MergeResult(
            clean_merges=clean_merges,
            added=added,
            removed=removed,
            conflicts=conflicts,
            errors=errors,
        )

    def _process_single_file(
        self,
        project_path: Path,
        old_fresh: Path,
        new_fresh: Path,
        relative_path: str,
        git_utils: GitUtils,
        dry_run: bool,
        three_way: bool = True,
    ) -> str:
        """Process a single file for merge.

        Returns:
            ``"clean"`` for clean merge, ``"conflict"`` for conflicts,
            or an error message string.
        """
        base_file = old_fresh / relative_path
        current_file = project_path / relative_path
        new_file = new_fresh / relative_path

        if not current_file.exists():
            # User deleted the file — skip
            return "clean"

        try:
            # Check for binary files
            if self._is_binary(current_file) or self._is_binary(new_file):
                base_bytes = base_file.read_bytes() if base_file.exists() else b""
                current_bytes = current_file.read_bytes()
                new_bytes = new_file.read_bytes()

                if current_bytes != base_bytes and new_bytes != base_bytes:
                    # Both sides modified a binary file
                    if not dry_run:
                        conflict_path = current_file.with_suffix(current_file.suffix + ".conflict")
                        conflict_path.write_bytes(new_bytes)
                    return "conflict"
                elif new_bytes != base_bytes:
                    if not dry_run:
                        current_file.write_bytes(new_bytes)
                    return "clean"
                return "clean"

            base_content = base_file.read_text(encoding="utf-8") if base_file.exists() else ""
            current_content = current_file.read_text(encoding="utf-8")
            new_content = new_file.read_text(encoding="utf-8")

            # Check which sides changed
            user_changed = current_content != base_content
            template_changed = new_content != base_content

            if not user_changed and not template_changed:
                return "clean"
            elif user_changed and not template_changed:
                return "clean"
            elif not user_changed and template_changed:
                if not dry_run:
                    current_file.write_text(new_content, encoding="utf-8")
                return "clean"
            else:
                # Both changed — 3-way merge
                merged, has_conflicts = git_utils.merge_file(base_content, current_content, new_content)
                if not dry_run:
                    current_file.write_text(merged, encoding="utf-8")
                return "conflict" if has_conflicts else "clean"

        except Exception as exc:
            return f"Failed to merge {relative_path}: {exc}"

    def _is_binary(self, path: Path) -> bool:
        """Heuristic check for binary files."""
        try:
            chunk = path.read_bytes()[:8192]
            return b"\x00" in chunk
        except Exception:  # nosec B110
            logger.debug("Failed binary check for %s", path, exc_info=True)
            return False

check_update_available(project_path)

Check if an update is available for the project.

Parameters:

Name Type Description Default
project_path Path

Path to the project.

 required

Returns:

Type Description
str | None

Latest version string if an update is available, None
str | None

otherwise.
Source code in src/nskit/client/update.py 
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
def check_update_available(self, project_path: Path) -> str | None:
    """Check if an update is available for the project.

    Args:
        project_path: Path to the project.

    Returns:
        Latest version string if an update is available, ``None``
        otherwise.
    """
    config_mgr = ConfigManager(project_path, self.config_dir, self.config_filename)
    if not config_mgr.is_recipe_based():
        return None

    try:
        config = config_mgr.load_config()
        if config.metadata is None:
            return None

        resolver = VersionResolver(self.backend)
        current_image = config.metadata.docker_image
        # Extract current version from image tag
        current_version = current_image.rsplit(":", 1)[-1] if ":" in current_image else "latest"
        update_needed, resolved = resolver.check_update_needed(config.metadata.recipe_name, current_version)
        if update_needed:
            return resolved
    except Exception:  # nosec B110
        logger.warning("Could not check for updates — run with debug logging for details")
        logger.debug("Update check failed for %s", project_path, exc_info=True)

    return None

update_project(project_path, target_version, diff_mode=DiffMode.THREE_WAY, dry_run=False)

Update project to target version.

Parameters:

Name Type Description Default
project_path Path

Path to the project.

 required
target_version str

Target version to update to.

 required
diff_mode DiffMode

Diff mode (2-way or 3-way).

 THREE_WAY
dry_run bool

If True, analyse changes without writing files.

 False

Returns:

Type Description
UpdateResult

Update result with conflicts and errors.
Source code in src/nskit/client/update.py 
 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
def update_project(
    self,
    project_path: Path,
    target_version: str,
    diff_mode: DiffMode = DiffMode.THREE_WAY,
    dry_run: bool = False,
) -> UpdateResult:
    """Update project to target version.

    Args:
        project_path: Path to the project.
        target_version: Target version to update to.
        diff_mode: Diff mode (2-way or 3-way).
        dry_run: If ``True``, analyse changes without writing files.

    Returns:
        Update result with conflicts and errors.
    """
    # Validate git status
    git_utils = GitUtils(project_path)
    if not git_utils.is_git_repository():
        raise GitStatusError("Project is not a git repository")

    if git_utils.has_uncommitted_changes():
        raise GitStatusError("Project has uncommitted changes. Commit or stash them first.")

    config_mgr = ConfigManager(project_path, self.config_dir, self.config_filename)
    config = config_mgr.load_config()

    if config.metadata is None:
        return UpdateResult(
            success=False,
            errors=["Recipe configuration missing metadata"],
        )

    if self.engine is None:
        return UpdateResult(
            success=False,
            errors=["No recipe engine configured for project generation"],
        )

    generator = ProjectGenerator(self.backend, self.engine)
    file_discovery = FileDiscovery()
    diff_engine = DiffEngine(file_discovery=file_discovery)

    old_fresh = None
    new_fresh = None
    try:
        _, old_fresh, new_fresh = generator.generate_project_states(config, target_version, project_path, diff_mode)

        merge_result = self._extract_and_process_changes(
            project_path=project_path,
            old_fresh=old_fresh,
            new_fresh=new_fresh,
            diff_engine=diff_engine,
            git_utils=git_utils,
            diff_mode=diff_mode,
            dry_run=dry_run,
        )

        result = UpdateResult(
            success=len(merge_result.errors) == 0,
            files_updated=merge_result.clean_merges,
            files_added=merge_result.added,
            files_removed=merge_result.removed,
            files_with_conflicts=merge_result.conflicts,
            clean_merges=merge_result.clean_merges,
            errors=merge_result.errors,
        )

        if result.success and not dry_run:
            config_mgr.update_config_version(target_version, config.metadata.recipe_name)

        return result

    except UpdateError:
        raise
    except Exception as exc:
        return UpdateResult(success=False, errors=[f"Update failed: {exc}"])
    finally:
        paths_to_clean = [p for p in [old_fresh, new_fresh] if p is not None]
        if paths_to_clean:
            generator.cleanup_states(*paths_to_clean)

UpdateResult

Bases: BaseModel

Result of recipe update.

Source code in src/nskit/client/models.py 
31
32
33
34
35
36
37
38
39
40
41
class UpdateResult(BaseModel):
    """Result of recipe update."""

    success: bool
    files_updated: list[str] = Field(default_factory=list)
    files_added: list[str] = Field(default_factory=list)
    files_removed: list[str] = Field(default_factory=list)
    files_with_conflicts: list[str] = Field(default_factory=list)
    clean_merges: list[str] = Field(default_factory=list)
    errors: list[str] = Field(default_factory=list)
    warnings: list[str] = Field(default_factory=list)

nskit.recipes.recipe.RecipeRecipe

Bases: PyRecipe

A Recipe for creating Recipes, meta!

Source code in src/nskit/recipes/recipe.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class RecipeRecipe(PyRecipe):
    """A Recipe for creating Recipes, meta!"""

    recipe_entrypoint: str = RECIPE_ENTRYPOINT
    contents: list[Union[File, Folder]] = Field(
        [
            ingredients.gitignore,
            ingredients.noxfile,
            ingredients.pre_commit,
            recipe_ingredients.pyproject_toml,
            recipe_ingredients.readme_md,
            recipe_ingredients.dockerfile,
            recipe_ingredients.docker_ignore,
            ingredients.test_dir,
            recipe_ingredients.src_dir,
            ingredients.docs_dir,
            LicenseFile(),
        ],
        description="The folder contents",
    )

contents = Field([ingredients.gitignore, ingredients.noxfile, ingredients.pre_commit, recipe_ingredients.pyproject_toml, recipe_ingredients.readme_md, recipe_ingredients.dockerfile, recipe_ingredients.docker_ignore, ingredients.test_dir, recipe_ingredients.src_dir, ingredients.docs_dir, LicenseFile()], description='The folder contents') class-attribute instance-attribute

nskit.recipes.python.api.APIRecipe

Bases: PyRecipe

API Service Recipe.

Source code in src/nskit/recipes/python/api.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class APIRecipe(PyRecipe):
    """API Service Recipe."""

    contents: list[Union[File, Folder]] = Field(
        [
            ingredients.gitignore,
            ingredients.noxfile,
            ingredients.pre_commit,
            api_ingredients.pyproject_toml,
            api_ingredients.readme_md,
            ingredients.test_dir,
            api_ingredients.src_dir,
            api_ingredients.docker.docker_ignore,
            api_ingredients.docker.dockerfile,
            ingredients.docs_dir,
            LicenseFile(),
        ],
        description="The folder contents",
    )

contents = Field([ingredients.gitignore, ingredients.noxfile, ingredients.pre_commit, api_ingredients.pyproject_toml, api_ingredients.readme_md, ingredients.test_dir, api_ingredients.src_dir, api_ingredients.docker.docker_ignore, api_ingredients.docker.dockerfile, ingredients.docs_dir, LicenseFile()], description='The folder contents') class-attribute instance-attribute

nskit.recipes.python.package.PackageRecipe

Bases: PyRecipe

Package Recipe.

Source code in src/nskit/recipes/python/package.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class PackageRecipe(PyRecipe):
    """Package Recipe."""

    contents: list[Union[File, Folder]] = Field(
        [
            ingredients.gitignore,
            ingredients.noxfile,
            ingredients.pre_commit,
            ingredients.pyproject_toml,
            ingredients.readme_md,
            ingredients.test_dir,
            ingredients.src_dir,
            ingredients.docs_dir,
            LicenseFile(),
        ],
        description="The folder contents",
    )

contents = Field([ingredients.gitignore, ingredients.noxfile, ingredients.pre_commit, ingredients.pyproject_toml, ingredients.readme_md, ingredients.test_dir, ingredients.src_dir, ingredients.docs_dir, LicenseFile()], description='The folder contents') class-attribute instance-attribute

Python Ingredients

nskit.recipes.python.PyRecipe

Bases: CodeRecipe

Base recipe for python recipes.

Source code in src/nskit/recipes/python/__init__.py 
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
class PyRecipe(CodeRecipe):
    """Base recipe for python recipes."""

    version: str = __version__
    """Version of the recipe."""

    repo: PyRepoMetadata = Field(...)
    """Python repo metadata."""

    language: str = "python"
    """The primary language of the repo."""

    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",
    )

    @staticmethod
    def _to_pep8(value):
        return str(value).lower().replace(" ", "_").replace("-", "_")

    def model_post_init(self, *args):  # noqa: U100
        """Set repo name handling."""
        self.repo.name = self.name

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

version = __version__ class-attribute instance-attribute

Version of the recipe.

repo = Field(...) class-attribute instance-attribute

Python repo metadata.

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([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') class-attribute instance-attribute

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

git = GitConfig() class-attribute instance-attribute

context property

Get the context on the initialised recipe.

recipe_batch property

Get information about the specific info of this recipe.

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 
204
205
206
207
208
209
210
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 
212
213
214
215
216
217
218
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)

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 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
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]}

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 
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
@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

nskit.recipes.python.PyRepoMetadata

Bases: RepoMetadata

Repo Metadata for python templates.

Source code in src/nskit/recipes/python/__init__.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
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class PyRepoMetadata(RepoMetadata):
    """Repo Metadata for python templates."""

    _name: str = None

    @property
    def name(self):
        """Get repo name."""
        return self._name

    @name.setter
    def name(self, value):
        """Set repo name."""
        if isinstance(value, str):
            self._name = value

    def _get_name_parts(self):
        return re.split("|".join(map(re.escape, list(set(_DELIMITERS + [self.repo_separator])))), self.name)

    @property
    def py_name(self):
        """Get python module name."""
        return ".".join(self._get_name_parts())

    @property
    def py_root(self):
        """Get root python module name."""
        return self._get_name_parts()[0]

    @property
    def src_path(self):
        """Get module folder structure (src not included)."""
        return Path(*self._get_name_parts()).as_posix()

    @property
    def module_depth(self) -> int:
        """Get the module depth.

        ``a.b.c`` has a depth of 3, ``a`` has a depth of 1
        """
        return len(self._get_name_parts())

owner = Field(..., description='Who is the owner of the repo') class-attribute instance-attribute

email = Field(..., description='The email for the repo owner') class-attribute instance-attribute

url = Field(..., description='The Repository url.') class-attribute instance-attribute

name property writable

Get repo name.

py_name property

Get python module name.

py_root property

Get root python module name.

src_path property

Get module folder structure (src not included).

module_depth property

Get the module depth.

a.b.c has a depth of 3, a has a depth of 1

nskit.recipes.python.ingredients

Ingredients for repos.

gitignore = File(name='.gitignore', content='nskit.recipes.python.ingredients.tools:gitignore.jinja') module-attribute

noxfile = File(name='noxfile.py', content='nskit.recipes.python.ingredients.tools:noxfile.py.jinja') module-attribute

pre_commit = File(name='.pre-commit-config.yaml', content='nskit.recipes.python.ingredients.tools:pre-commit-config.yaml.jinja') module-attribute

pyproject_toml = File(name='pyproject.toml', content='nskit.recipes.python.ingredients.tools:pyproject.toml.jinja') module-attribute

readme_md = File(name='README.md', content='nskit.recipes.python.ingredients.tools:README.md.jinja') module-attribute

test_dir = Folder(name='tests', contents=[Folder(name='unit', contents=[File(name='test__version.py', content=test_version)]), Folder(name='functional', contents=[File(name=_GIT_KEEP, content='')]), Folder(name='integration', contents=[File(name=_GIT_KEEP, content='')]), Folder(name='performance', contents=[File(name=_GIT_KEEP, content='')]), Folder(name='smoke', contents=[File(name=_GIT_KEEP, content='')])]) module-attribute

src_dir = Folder(name='src', contents=[Folder(id_='src_path', name='{{repo.src_path}}', contents=[File(name='__init__.py', content='nskit.recipes.python.ingredients.src:__init__.py.jinja'), File(name='_version.py', content='__version__ = "0.0.0"\n')])]) module-attribute

nskit.recipes.python.ingredients.api

API Service ingredients.

Contains fastapi based api service ingredients.

pyproject_toml = File(name='pyproject.toml', content='nskit.recipes.python.ingredients.api:pyproject.toml.jinja') module-attribute

readme_md = File(name='README.md', content='nskit.recipes.python.ingredients.api:README.md.jinja') module-attribute

src_dir

Adds app.py, server.py, api/__init__.py, base.py to [nskit.recipes.python.ingredients.src_dir]

nskit.recipes.python.ingredients.docker

dockerfile = File(name='Dockerfile', content='nskit.recipes.python.ingredients.docker:dockerfile.jinja') module-attribute

docker_ignore = File(name='.dockerignore', content='nskit.recipes.python.ingredients.docker:dockerignore.jinja') module-attribute

nskit.recipes.python.ingredients.recipe

Ingredients for a recipe recipe.

pyproject_toml = File(name='pyproject.toml', content='nskit.recipes.python.ingredients.recipe:pyproject.toml.jinja') module-attribute

readme_md = File(name='README.md', content='nskit.recipes.python.ingredients.recipe:readme.md.jinja') module-attribute

src_dir

Adds recipe.py and ingredient.py.jinja to [nskit.recipes.python.ingredients.src_dir]