Client module for recipe operations.

DiscoveryClient

Client for discovering available recipes.

Source code in src/nskit/client/discovery.py

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

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

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

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

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

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

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

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

LocalEngine

Bases: RecipeEngine

Execute recipes from locally installed packages.

Source code in src/nskit/client/engines/local.py

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

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

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

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

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

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

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

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

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

RecipeEngine

Bases: ABC

Abstract interface for recipe execution engines.

Source code in src/nskit/client/engines/base.py

class RecipeEngine(ABC):
    """Abstract interface for recipe execution engines."""

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

        Args:
            recipe: Recipe name
            version: Recipe version
            parameters: Recipe parameters
            output_dir: Output directory
            image_url: Docker image URL (for Docker engine)
            entrypoint: Recipe entrypoint (for Local engine)

        Returns:
            Recipe execution result
        """
        pass

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

Execute 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	required
image_url	str	Docker image URL (for Docker engine)	None
entrypoint	str	Recipe entrypoint (for Local engine)	None

Returns:

Type	Description
RecipeResult	Recipe execution result

Source code in src/nskit/client/engines/base.py

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

    Args:
        recipe: Recipe name
        version: Recipe version
        parameters: Recipe parameters
        output_dir: Output directory
        image_url: Docker image URL (for Docker engine)
        entrypoint: Recipe entrypoint (for Local engine)

    Returns:
        Recipe execution result
    """
    pass

RecipeInfo

Bases: BaseModel

Information about a recipe.

Source code in src/nskit/client/models.py

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

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)

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

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

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

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

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.client.recipes.RecipeClient

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

Source code in src/nskit/client/recipes.py

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}"

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

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

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

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

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

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

nskit.client.update.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

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

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

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)

nskit.client.discovery.DiscoveryClient

Client for discovering available recipes.

Source code in src/nskit/client/discovery.py

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)

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

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

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

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)

Engines

nskit.client.engines.base.RecipeEngine

Bases: ABC

Abstract interface for recipe execution engines.

Source code in src/nskit/client/engines/base.py

class RecipeEngine(ABC):
    """Abstract interface for recipe execution engines."""

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

        Args:
            recipe: Recipe name
            version: Recipe version
            parameters: Recipe parameters
            output_dir: Output directory
            image_url: Docker image URL (for Docker engine)
            entrypoint: Recipe entrypoint (for Local engine)

        Returns:
            Recipe execution result
        """
        pass

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

Execute 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	required
image_url	str	Docker image URL (for Docker engine)	None
entrypoint	str	Recipe entrypoint (for Local engine)	None

Returns:

Type	Description
RecipeResult	Recipe execution result

Source code in src/nskit/client/engines/base.py

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

    Args:
        recipe: Recipe name
        version: Recipe version
        parameters: Recipe parameters
        output_dir: Output directory
        image_url: Docker image URL (for Docker engine)
        entrypoint: Recipe entrypoint (for Local engine)

    Returns:
        Recipe execution result
    """
    pass

nskit.client.engines.local.LocalEngine

Bases: RecipeEngine

Execute recipes from locally installed packages.

Source code in src/nskit/client/engines/local.py

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

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

nskit.client.engines.docker.DockerEngine

Bases: RecipeEngine

Execute recipes in Docker containers.

Source code in src/nskit/client/engines/docker.py

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

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

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

Backends

nskit.client.backends.base.RecipeBackend

Bases: ABC

Abstract interface for recipe backends.

Source code in src/nskit/client/backends/base.py

class RecipeBackend(ABC):
    """Abstract interface for recipe backends."""

    @property
    @abstractmethod
    def entrypoint(self) -> str:
        """Get the recipe entrypoint for this backend."""
        pass

    @abstractmethod
    def list_recipes(self) -> list[RecipeInfo]:
        """List all available recipes.

        Returns:
            List of recipe information.
        """
        pass

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

        Args:
            recipe: Recipe name.

        Returns:
            List of version strings.
        """
        pass

    @abstractmethod
    def fetch_recipe(self, recipe: str, version: str, dest: Path) -> Path:
        """Fetch a recipe to a destination directory.

        Args:
            recipe: Recipe name.
            version: Recipe version.
            dest: Destination directory.

        Returns:
            Path to the fetched recipe.
        """
        pass

    def get_recipe_metadata(self, recipe: str, version: str) -> dict[str, Any]:
        """Get metadata for a specific recipe version.

        Args:
            recipe: Recipe name.
            version: Recipe version.

        Returns:
            Recipe metadata dictionary.
        """
        return {}

    def get_image_url(self, recipe: str, version: str) -> str:
        """Get Docker image URL for a recipe.

        Args:
            recipe: Recipe name.
            version: Recipe version.

        Returns:
            Docker image URL.
        """
        raise NotImplementedError("This backend does not support Docker images")

    def pull_image(self, image_url: str) -> None:
        """Pull Docker image.

        Args:
            image_url: Docker image URL to pull.
        """
        raise NotImplementedError("This backend does not support Docker images")

entrypoint abstractmethod property

Get the recipe entrypoint for this backend.

fetch_recipe(recipe, version, dest) abstractmethod

Fetch a recipe to a destination directory.

Parameters:

Name	Type	Description	Default
recipe	str	Recipe name.	required
version	str	Recipe version.	required
dest	Path	Destination directory.	required

Returns:

Type	Description
Path	Path to the fetched recipe.

Source code in src/nskit/client/backends/base.py

@abstractmethod
def fetch_recipe(self, recipe: str, version: str, dest: Path) -> Path:
    """Fetch a recipe to a destination directory.

    Args:
        recipe: Recipe name.
        version: Recipe version.
        dest: Destination directory.

    Returns:
        Path to the fetched recipe.
    """
    pass

get_image_url(recipe, version)

Get Docker image URL for a recipe.

Parameters:

Name	Type	Description	Default
recipe	str	Recipe name.	required
version	str	Recipe version.	required

Returns:

Type	Description
str	Docker image URL.

Source code in src/nskit/client/backends/base.py

def get_image_url(self, recipe: str, version: str) -> str:
    """Get Docker image URL for a recipe.

    Args:
        recipe: Recipe name.
        version: Recipe version.

    Returns:
        Docker image URL.
    """
    raise NotImplementedError("This backend does not support Docker images")

get_recipe_metadata(recipe, version)

Get metadata for a specific recipe version.

Parameters:

Name	Type	Description	Default
recipe	str	Recipe name.	required
version	str	Recipe version.	required

Returns:

Type	Description
dict[str, Any]	Recipe metadata dictionary.

Source code in src/nskit/client/backends/base.py

def get_recipe_metadata(self, recipe: str, version: str) -> dict[str, Any]:
    """Get metadata for a specific recipe version.

    Args:
        recipe: Recipe name.
        version: Recipe version.

    Returns:
        Recipe metadata dictionary.
    """
    return {}

get_recipe_versions(recipe) abstractmethod

Get available versions for a recipe.

Parameters:

Name	Type	Description	Default
recipe	str	Recipe name.	required

Returns:

Type	Description
list[str]	List of version strings.

Source code in src/nskit/client/backends/base.py

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

    Args:
        recipe: Recipe name.

    Returns:
        List of version strings.
    """
    pass

list_recipes() abstractmethod

List all available recipes.

Returns:

Type	Description
list[RecipeInfo]	List of recipe information.

Source code in src/nskit/client/backends/base.py

@abstractmethod
def list_recipes(self) -> list[RecipeInfo]:
    """List all available recipes.

    Returns:
        List of recipe information.
    """
    pass

pull_image(image_url)

Pull Docker image.

Parameters:

Name	Type	Description	Default
image_url	str	Docker image URL to pull.	required

Source code in src/nskit/client/backends/base.py

def pull_image(self, image_url: str) -> None:
    """Pull Docker image.

    Args:
        image_url: Docker image URL to pull.
    """
    raise NotImplementedError("This backend does not support Docker images")

nskit.client.backends.local.LocalBackend

Bases: RecipeBackend

Backend for local filesystem recipes.

Source code in src/nskit/client/backends/local.py

class LocalBackend(RecipeBackend):
    """Backend for local filesystem recipes."""

    def __init__(self, recipes_dir: Path, entrypoint: str = RECIPE_ENTRYPOINT):
        """Initialize local backend.

        Args:
            recipes_dir: Directory containing recipes.
            entrypoint: Recipe entrypoint name.
        """
        self.recipes_dir = Path(recipes_dir)
        self._entrypoint = entrypoint

    @property
    def entrypoint(self) -> str:
        """Get the recipe entrypoint."""
        return self._entrypoint

    def list_recipes(self) -> list[RecipeInfo]:
        """List recipes from local directory.

        Returns:
            List of recipe information found in the recipes directory.
        """
        recipes = []
        if not self.recipes_dir.exists():
            return recipes

        for recipe_dir in self.recipes_dir.iterdir():
            if recipe_dir.is_dir() and not recipe_dir.name.startswith("."):
                versions = []
                for version_dir in recipe_dir.iterdir():
                    if version_dir.is_dir() and not version_dir.name.startswith("."):
                        versions.append(version_dir.name)
                if versions:
                    recipes.append(
                        RecipeInfo(
                            name=recipe_dir.name,
                            versions=sorted(versions),
                            description=f"Local recipe: {recipe_dir.name}",
                        )
                    )
        return recipes

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

        Args:
            recipe: Recipe name.

        Returns:
            Sorted list of version strings.
        """
        recipe_dir = self.recipes_dir / recipe
        if not recipe_dir.exists():
            return []
        versions = []
        for version_dir in recipe_dir.iterdir():
            if version_dir.is_dir() and not version_dir.name.startswith("."):
                versions.append(version_dir.name)
        return sorted(versions)

    def fetch_recipe(self, recipe: str, version: str, dest: Path) -> Path:
        """Copy recipe from local directory.

        Args:
            recipe: Recipe name.
            version: Recipe version.
            dest: Destination directory.

        Returns:
            Path to the copied recipe.

        Raises:
            FileNotFoundError: If the recipe version does not exist.
        """
        import shutil

        source = self.recipes_dir / recipe / version
        if not source.exists():
            raise FileNotFoundError(f"Recipe {recipe} version {version} not found at {source}")
        dest.mkdir(parents=True, exist_ok=True)
        shutil.copytree(source, dest / recipe, dirs_exist_ok=True)
        return dest / recipe

entrypoint property

Get the recipe entrypoint.

__init__(recipes_dir, entrypoint=RECIPE_ENTRYPOINT)

Initialize local backend.

Parameters:

Name	Type	Description	Default
recipes_dir	Path	Directory containing recipes.	required
entrypoint	str	Recipe entrypoint name.	RECIPE_ENTRYPOINT

Source code in src/nskit/client/backends/local.py

def __init__(self, recipes_dir: Path, entrypoint: str = RECIPE_ENTRYPOINT):
    """Initialize local backend.

    Args:
        recipes_dir: Directory containing recipes.
        entrypoint: Recipe entrypoint name.
    """
    self.recipes_dir = Path(recipes_dir)
    self._entrypoint = entrypoint

fetch_recipe(recipe, version, dest)

Copy recipe from local directory.

Parameters:

Name	Type	Description	Default
recipe	str	Recipe name.	required
version	str	Recipe version.	required
dest	Path	Destination directory.	required

Returns:

Type	Description
Path	Path to the copied recipe.

Raises:

Type	Description
FileNotFoundError	If the recipe version does not exist.

Source code in src/nskit/client/backends/local.py

def fetch_recipe(self, recipe: str, version: str, dest: Path) -> Path:
    """Copy recipe from local directory.

    Args:
        recipe: Recipe name.
        version: Recipe version.
        dest: Destination directory.

    Returns:
        Path to the copied recipe.

    Raises:
        FileNotFoundError: If the recipe version does not exist.
    """
    import shutil

    source = self.recipes_dir / recipe / version
    if not source.exists():
        raise FileNotFoundError(f"Recipe {recipe} version {version} not found at {source}")
    dest.mkdir(parents=True, exist_ok=True)
    shutil.copytree(source, dest / recipe, dirs_exist_ok=True)
    return dest / recipe

get_recipe_versions(recipe)

Get versions for a recipe.

Parameters:

Name	Type	Description	Default
recipe	str	Recipe name.	required

Returns:

Type	Description
list[str]	Sorted list of version strings.

Source code in src/nskit/client/backends/local.py

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

    Args:
        recipe: Recipe name.

    Returns:
        Sorted list of version strings.
    """
    recipe_dir = self.recipes_dir / recipe
    if not recipe_dir.exists():
        return []
    versions = []
    for version_dir in recipe_dir.iterdir():
        if version_dir.is_dir() and not version_dir.name.startswith("."):
            versions.append(version_dir.name)
    return sorted(versions)

list_recipes()

List recipes from local directory.

Returns:

Type	Description
list[RecipeInfo]	List of recipe information found in the recipes directory.

Source code in src/nskit/client/backends/local.py

def list_recipes(self) -> list[RecipeInfo]:
    """List recipes from local directory.

    Returns:
        List of recipe information found in the recipes directory.
    """
    recipes = []
    if not self.recipes_dir.exists():
        return recipes

    for recipe_dir in self.recipes_dir.iterdir():
        if recipe_dir.is_dir() and not recipe_dir.name.startswith("."):
            versions = []
            for version_dir in recipe_dir.iterdir():
                if version_dir.is_dir() and not version_dir.name.startswith("."):
                    versions.append(version_dir.name)
            if versions:
                recipes.append(
                    RecipeInfo(
                        name=recipe_dir.name,
                        versions=sorted(versions),
                        description=f"Local recipe: {recipe_dir.name}",
                    )
                )
    return recipes

nskit.client.backends.github.GitHubBackend

Bases: RecipeBackend

Backend that fetches recipes from GitHub releases.

Source code in src/nskit/client/backends/github.py

class GitHubBackend(RecipeBackend):
    """Backend that fetches recipes from GitHub releases."""

    def __init__(
        self,
        org: str,
        repo_pattern: str = "{recipe_name}",
        token: str | SecretStr | None = None,
        entrypoint: str = RECIPE_ENTRYPOINT,
    ):
        """Initialize GitHub backend.

        Args:
            org: GitHub organization.
            repo_pattern: Pattern for repository names (use {recipe_name} placeholder).
            token: Optional GitHub token (str or SecretStr; uses gh CLI if not provided).
            entrypoint: Recipe entrypoint name.
        """
        self.org = org
        self.repo_pattern = repo_pattern
        self._token: SecretStr | None = SecretStr(token) if isinstance(token, str) else token
        self._github: GhApi | None = None
        self._entrypoint = entrypoint
        if GhApi is None:
            raise ImportError("GitHubBackend requires ghapi. Install with: pip install nskit[github]")

    @property
    def entrypoint(self) -> str:
        """Get the recipe entrypoint."""
        return self._entrypoint

    def _get_token(self) -> str:
        """Get GitHub token from gh CLI or provided token.

        Returns:
            GitHub token string.

        Raises:
            RuntimeError: If gh CLI is not found or not authenticated.
        """
        if self._token:
            return self._token.get_secret_value()
        try:
            result = subprocess.run(  # nosec B603, B607
                ["gh", "auth", "token"], check=True, capture_output=True, text=True
            )
            self._token = SecretStr(result.stdout.strip())
            return self._token.get_secret_value()
        except FileNotFoundError:
            raise RuntimeError("GitHub CLI (gh) not found. Please install it: https://cli.github.com/") from None
        except subprocess.CalledProcessError:
            raise RuntimeError("GitHub authentication required. Please run: gh auth login") from None

    def _get_client(self) -> GhApi:
        """Get authenticated GitHub client.

        Returns:
            Authenticated ``GhApi`` instance.
        """
        if self._github is None:
            self._github = GhApi(token=self._get_token())
        return self._github

    def _get_repo_name(self, recipe_name: str) -> str:
        """Build repository name from pattern.

        Args:
            recipe_name: Recipe name.

        Returns:
            Repository name string.
        """
        validate_recipe_name(recipe_name)
        return self.repo_pattern.format(recipe_name=recipe_name)

    def list_recipes(self) -> list[RecipeInfo]:
        """List available recipes from GitHub org.

        Returns:
            List of recipe information from the organization's repositories.
        """
        from nskit.client.backends.image_labels import get_recipe_name, read_remote_labels

        github = self._get_client()
        repos = github.repos.list_for_org(self.org, type="public")

        recipes = []
        for repo in repos:
            try:
                releases = github.repos.list_releases(self.org, repo.name)
                versions = [r.tag_name for r in releases if not r.draft]
            except Exception:
                logger.warning("Could not fetch releases for %s/%s — skipping", self.org, repo.name)
                versions = []

            name = repo.name
            if versions:
                image_url = self.get_image_url(name, versions[0])
                try:
                    labels = read_remote_labels(image_url, token=self._get_token())
                    name = get_recipe_name(labels) or name
                except Exception:
                    logger.debug("Failed to read labels for %s", image_url, exc_info=True)

            recipes.append(RecipeInfo(name=name, versions=versions, description=repo.description))
        return recipes

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

        Args:
            recipe_name: Recipe name.

        Returns:
            List of version tag strings.
        """
        github = self._get_client()
        repo_name = self._get_repo_name(recipe_name)
        try:
            releases = github.repos.list_releases(self.org, repo_name)
            return [r.tag_name for r in releases if not r.draft]
        except Exception:
            logger.warning("Could not fetch versions for recipe '%s'", recipe_name)
            return []

    def fetch_recipe(self, recipe_name: str, version: str, target_path: Path) -> Path:
        """Fetch recipe from GitHub release.

        Args:
            recipe_name: Recipe name.
            version: Recipe version (tag name).
            target_path: Where to extract recipe.

        Returns:
            Path to extracted recipe.
        """
        github = self._get_client()
        repo_name = self._get_repo_name(recipe_name)
        github.repos.get_release_by_tag(self.org, repo_name, version)

        archive_url = f"https://github.com/{self.org}/{repo_name}/archive/refs/tags/{version}.zip"

        with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as tmp:
            subprocess.run(  # nosec B603, B607
                ["curl", "-L", "-o", tmp.name, archive_url], check=True, capture_output=True
            )
            # Extract archive (with zip-slip protection)
            with zipfile.ZipFile(tmp.name, "r") as zip_ref:
                for member in zip_ref.namelist():
                    member_path = (target_path / member).resolve()
                    if not str(member_path).startswith(str(target_path.resolve())):
                        raise ValueError(f"Zip entry {member!r} would escape target directory")
                zip_ref.extractall(target_path)

        # GitHub archives extract to {repo}-{tag}/ directory
        extracted_dir = target_path / f"{repo_name}-{version.lstrip('v')}"
        if not extracted_dir.exists():
            extracted_dir = target_path / f"{repo_name}-{version}"
        return extracted_dir

    def get_image_url(self, recipe: str, version: str) -> str:
        """Get Docker image URL from GitHub Container Registry.

        Args:
            recipe: Recipe name.
            version: Recipe version.

        Returns:
            Docker image URL.
        """
        repo_name = self._get_repo_name(recipe)
        validate_version(version)
        return f"ghcr.io/{self.org}/{repo_name}:{version}"

    def pull_image(self, image_url: str) -> None:
        """Pull Docker image from GitHub Container Registry.

        Args:
            image_url: Docker image URL to pull.
        """
        validate_image_url(image_url)
        token = self._get_token()
        subprocess.run(  # nosec B603, B607
            ["docker", "login", "ghcr.io", "-u", "token", "--password-stdin"],
            input=token,
            text=True,
            check=True,
            capture_output=True,
        )
        subprocess.run(["docker", "pull", image_url], check=True, capture_output=True)  # nosec B603, B607

entrypoint property

Get the recipe entrypoint.

__init__(org, repo_pattern='{recipe_name}', token=None, entrypoint=RECIPE_ENTRYPOINT)

Initialize GitHub backend.

Parameters:

Name	Type	Description	Default
org	str	GitHub organization.	required
repo_pattern	str	Pattern for repository names (use {recipe_name} placeholder).	'{recipe_name}'
token	str | SecretStr | None	Optional GitHub token (str or SecretStr; uses gh CLI if not provided).	None
entrypoint	str	Recipe entrypoint name.	RECIPE_ENTRYPOINT

Source code in src/nskit/client/backends/github.py

def __init__(
    self,
    org: str,
    repo_pattern: str = "{recipe_name}",
    token: str | SecretStr | None = None,
    entrypoint: str = RECIPE_ENTRYPOINT,
):
    """Initialize GitHub backend.

    Args:
        org: GitHub organization.
        repo_pattern: Pattern for repository names (use {recipe_name} placeholder).
        token: Optional GitHub token (str or SecretStr; uses gh CLI if not provided).
        entrypoint: Recipe entrypoint name.
    """
    self.org = org
    self.repo_pattern = repo_pattern
    self._token: SecretStr | None = SecretStr(token) if isinstance(token, str) else token
    self._github: GhApi | None = None
    self._entrypoint = entrypoint
    if GhApi is None:
        raise ImportError("GitHubBackend requires ghapi. Install with: pip install nskit[github]")

fetch_recipe(recipe_name, version, target_path)

Fetch recipe from GitHub release.

Parameters:

Name	Type	Description	Default
recipe_name	str	Recipe name.	required
version	str	Recipe version (tag name).	required
target_path	Path	Where to extract recipe.	required

Returns:

Type	Description
Path	Path to extracted recipe.

Source code in src/nskit/client/backends/github.py

def fetch_recipe(self, recipe_name: str, version: str, target_path: Path) -> Path:
    """Fetch recipe from GitHub release.

    Args:
        recipe_name: Recipe name.
        version: Recipe version (tag name).
        target_path: Where to extract recipe.

    Returns:
        Path to extracted recipe.
    """
    github = self._get_client()
    repo_name = self._get_repo_name(recipe_name)
    github.repos.get_release_by_tag(self.org, repo_name, version)

    archive_url = f"https://github.com/{self.org}/{repo_name}/archive/refs/tags/{version}.zip"

    with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as tmp:
        subprocess.run(  # nosec B603, B607
            ["curl", "-L", "-o", tmp.name, archive_url], check=True, capture_output=True
        )
        # Extract archive (with zip-slip protection)
        with zipfile.ZipFile(tmp.name, "r") as zip_ref:
            for member in zip_ref.namelist():
                member_path = (target_path / member).resolve()
                if not str(member_path).startswith(str(target_path.resolve())):
                    raise ValueError(f"Zip entry {member!r} would escape target directory")
            zip_ref.extractall(target_path)

    # GitHub archives extract to {repo}-{tag}/ directory
    extracted_dir = target_path / f"{repo_name}-{version.lstrip('v')}"
    if not extracted_dir.exists():
        extracted_dir = target_path / f"{repo_name}-{version}"
    return extracted_dir

get_image_url(recipe, version)

Get Docker image URL from GitHub Container Registry.

Parameters:

Name	Type	Description	Default
recipe	str	Recipe name.	required
version	str	Recipe version.	required

Returns:

Type	Description
str	Docker image URL.

Source code in src/nskit/client/backends/github.py

def get_image_url(self, recipe: str, version: str) -> str:
    """Get Docker image URL from GitHub Container Registry.

    Args:
        recipe: Recipe name.
        version: Recipe version.

    Returns:
        Docker image URL.
    """
    repo_name = self._get_repo_name(recipe)
    validate_version(version)
    return f"ghcr.io/{self.org}/{repo_name}:{version}"

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 version tag strings.

Source code in src/nskit/client/backends/github.py

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

    Args:
        recipe_name: Recipe name.

    Returns:
        List of version tag strings.
    """
    github = self._get_client()
    repo_name = self._get_repo_name(recipe_name)
    try:
        releases = github.repos.list_releases(self.org, repo_name)
        return [r.tag_name for r in releases if not r.draft]
    except Exception:
        logger.warning("Could not fetch versions for recipe '%s'", recipe_name)
        return []

list_recipes()

List available recipes from GitHub org.

Returns:

Type	Description
list[RecipeInfo]	List of recipe information from the organization's repositories.

Source code in src/nskit/client/backends/github.py

def list_recipes(self) -> list[RecipeInfo]:
    """List available recipes from GitHub org.

    Returns:
        List of recipe information from the organization's repositories.
    """
    from nskit.client.backends.image_labels import get_recipe_name, read_remote_labels

    github = self._get_client()
    repos = github.repos.list_for_org(self.org, type="public")

    recipes = []
    for repo in repos:
        try:
            releases = github.repos.list_releases(self.org, repo.name)
            versions = [r.tag_name for r in releases if not r.draft]
        except Exception:
            logger.warning("Could not fetch releases for %s/%s — skipping", self.org, repo.name)
            versions = []

        name = repo.name
        if versions:
            image_url = self.get_image_url(name, versions[0])
            try:
                labels = read_remote_labels(image_url, token=self._get_token())
                name = get_recipe_name(labels) or name
            except Exception:
                logger.debug("Failed to read labels for %s", image_url, exc_info=True)

        recipes.append(RecipeInfo(name=name, versions=versions, description=repo.description))
    return recipes

pull_image(image_url)

Pull Docker image from GitHub Container Registry.

Parameters:

Name	Type	Description	Default
image_url	str	Docker image URL to pull.	required

Source code in src/nskit/client/backends/github.py

def pull_image(self, image_url: str) -> None:
    """Pull Docker image from GitHub Container Registry.

    Args:
        image_url: Docker image URL to pull.
    """
    validate_image_url(image_url)
    token = self._get_token()
    subprocess.run(  # nosec B603, B607
        ["docker", "login", "ghcr.io", "-u", "token", "--password-stdin"],
        input=token,
        text=True,
        check=True,
        capture_output=True,
    )
    subprocess.run(["docker", "pull", image_url], check=True, capture_output=True)  # nosec B603, B607

nskit.client.backends.docker.DockerBackend

Bases: RecipeBackend

Backend that fetches recipes from Docker registry.

Source code in src/nskit/client/backends/docker.py

class DockerBackend(RecipeBackend):
    """Backend that fetches recipes from Docker registry."""

    def __init__(
        self,
        registry_url: str = "ghcr.io",
        image_prefix: str = "",
        auth_token: str | SecretStr | None = None,
        entrypoint: str = RECIPE_ENTRYPOINT,
        timeouts: DockerTimeouts | None = None,
    ):
        """Initialize Docker backend.

        Args:
            registry_url: Docker registry URL (e.g., ghcr.io).
            image_prefix: Prefix for image names (e.g., org/project).
            auth_token: Optional authentication token (str or SecretStr).
            entrypoint: Recipe entrypoint name.
            timeouts: Timeout configuration for Docker operations.
        """
        self.registry_url = registry_url
        self.image_prefix = image_prefix
        self._auth_token = SecretStr(auth_token) if isinstance(auth_token, str) else auth_token
        self._entrypoint = entrypoint
        self.timeouts = timeouts or DockerTimeouts()
        self._check_docker()

    @property
    def entrypoint(self) -> str:
        """Get the recipe entrypoint."""
        return self._entrypoint

    def _check_docker(self) -> None:
        """Check if Docker is installed and running."""
        try:
            subprocess.run(["docker", "info"], check=True, capture_output=True, timeout=5)  # nosec B603, B607
        except FileNotFoundError:
            raise RuntimeError("Docker not found. Please install Docker: https://docs.docker.com/get-docker/") from None
        except subprocess.CalledProcessError:
            raise RuntimeError("Docker is not running. Please start Docker Desktop or the Docker daemon.") from None
        except subprocess.TimeoutExpired:
            raise RuntimeError("Docker is not responding. Please check if Docker is running properly.") from None

    def _build_image_url(self, recipe_name: str, version: str) -> str:
        """Build Docker image URL.

        Args:
            recipe_name: Recipe name.
            version: Recipe version.

        Returns:
            Full image URL string.
        """
        validate_recipe_name(recipe_name)
        validate_version(version)
        if self.image_prefix:
            return f"{self.registry_url}/{self.image_prefix}/{recipe_name}:{version}"
        return f"{self.registry_url}/{recipe_name}:{version}"

    def _authenticate(self) -> None:
        """Authenticate with Docker registry if token provided."""
        if not self._auth_token:
            return
        subprocess.run(  # nosec B603, B607
            ["docker", "login", self.registry_url, "-u", "token", "--password-stdin"],
            input=self._auth_token.get_secret_value(),
            text=True,
            check=True,
            capture_output=True,
        )

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

        Note: Docker registries don't provide easy listing.
        This is a limitation of the Docker backend.
        """
        return []

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

        Note: Requires external API or manifest inspection.
        Returns empty list as Docker doesn't provide easy version listing.
        """
        return []

    def fetch_recipe(self, recipe_name: str, version: str, target_path: Path) -> Path:
        """Fetch recipe from Docker image.

        Args:
            recipe_name: Recipe name.
            version: Recipe version.
            target_path: Where to extract recipe.

        Returns:
            Path to extracted recipe.
        """
        self._authenticate()
        image_url = self._build_image_url(recipe_name, version)

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

        result = subprocess.run(  # nosec B603, B607
            ["docker", "create", image_url], check=True, capture_output=True, text=True, timeout=self.timeouts.cmd
        )
        container_id = result.stdout.strip()

        try:
            subprocess.run(  # nosec B603, B607
                ["docker", "cp", f"{container_id}:/app/recipes/", str(target_path)],
                check=True,
                timeout=self.timeouts.file_copy,
            )
        finally:
            subprocess.run(  # nosec B603, B607
                ["docker", "rm", container_id], check=True, capture_output=True, timeout=self.timeouts.cmd
            )

        return target_path / recipe_name

    def get_image_url(self, recipe: str, version: str) -> str:
        """Get Docker image URL for a recipe.

        Args:
            recipe: Recipe name.
            version: Recipe version.

        Returns:
            Docker image URL.
        """
        return self._build_image_url(recipe, version)

    def pull_image(self, image_url: str) -> None:
        """Pull Docker image.

        Args:
            image_url: Docker image URL to pull.
        """
        validate_image_url(image_url)
        self._authenticate()
        subprocess.run(  # nosec B603, B607
            ["docker", "pull", image_url], check=True, capture_output=True, timeout=self.timeouts.pull
        )

entrypoint property

Get the recipe entrypoint.

__init__(registry_url='ghcr.io', image_prefix='', auth_token=None, entrypoint=RECIPE_ENTRYPOINT, timeouts=None)

Initialize Docker backend.

Parameters:

Name	Type	Description	Default
registry_url	str	Docker registry URL (e.g., ghcr.io).	'ghcr.io'
image_prefix	str	Prefix for image names (e.g., org/project).	''
auth_token	str | SecretStr | None	Optional authentication token (str or SecretStr).	None
entrypoint	str	Recipe entrypoint name.	RECIPE_ENTRYPOINT
timeouts	DockerTimeouts | None	Timeout configuration for Docker operations.	None

Source code in src/nskit/client/backends/docker.py

def __init__(
    self,
    registry_url: str = "ghcr.io",
    image_prefix: str = "",
    auth_token: str | SecretStr | None = None,
    entrypoint: str = RECIPE_ENTRYPOINT,
    timeouts: DockerTimeouts | None = None,
):
    """Initialize Docker backend.

    Args:
        registry_url: Docker registry URL (e.g., ghcr.io).
        image_prefix: Prefix for image names (e.g., org/project).
        auth_token: Optional authentication token (str or SecretStr).
        entrypoint: Recipe entrypoint name.
        timeouts: Timeout configuration for Docker operations.
    """
    self.registry_url = registry_url
    self.image_prefix = image_prefix
    self._auth_token = SecretStr(auth_token) if isinstance(auth_token, str) else auth_token
    self._entrypoint = entrypoint
    self.timeouts = timeouts or DockerTimeouts()
    self._check_docker()

fetch_recipe(recipe_name, version, target_path)

Fetch recipe from Docker image.

Parameters:

Name	Type	Description	Default
recipe_name	str	Recipe name.	required
version	str	Recipe version.	required
target_path	Path	Where to extract recipe.	required

Returns:

Type	Description
Path	Path to extracted recipe.

Source code in src/nskit/client/backends/docker.py

def fetch_recipe(self, recipe_name: str, version: str, target_path: Path) -> Path:
    """Fetch recipe from Docker image.

    Args:
        recipe_name: Recipe name.
        version: Recipe version.
        target_path: Where to extract recipe.

    Returns:
        Path to extracted recipe.
    """
    self._authenticate()
    image_url = self._build_image_url(recipe_name, version)

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

    result = subprocess.run(  # nosec B603, B607
        ["docker", "create", image_url], check=True, capture_output=True, text=True, timeout=self.timeouts.cmd
    )
    container_id = result.stdout.strip()

    try:
        subprocess.run(  # nosec B603, B607
            ["docker", "cp", f"{container_id}:/app/recipes/", str(target_path)],
            check=True,
            timeout=self.timeouts.file_copy,
        )
    finally:
        subprocess.run(  # nosec B603, B607
            ["docker", "rm", container_id], check=True, capture_output=True, timeout=self.timeouts.cmd
        )

    return target_path / recipe_name

get_image_url(recipe, version)

Get Docker image URL for a recipe.

Parameters:

Name	Type	Description	Default
recipe	str	Recipe name.	required
version	str	Recipe version.	required

Returns:

Type	Description
str	Docker image URL.

Source code in src/nskit/client/backends/docker.py

def get_image_url(self, recipe: str, version: str) -> str:
    """Get Docker image URL for a recipe.

    Args:
        recipe: Recipe name.
        version: Recipe version.

    Returns:
        Docker image URL.
    """
    return self._build_image_url(recipe, version)

get_recipe_versions(recipe_name)

Get available versions for a recipe.

Note: Requires external API or manifest inspection. Returns empty list as Docker doesn't provide easy version listing.

Source code in src/nskit/client/backends/docker.py

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

    Note: Requires external API or manifest inspection.
    Returns empty list as Docker doesn't provide easy version listing.
    """
    return []

list_recipes()

List available recipes.

Note: Docker registries don't provide easy listing. This is a limitation of the Docker backend.

Source code in src/nskit/client/backends/docker.py

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

    Note: Docker registries don't provide easy listing.
    This is a limitation of the Docker backend.
    """
    return []

pull_image(image_url)

Pull Docker image.

Parameters:

Name	Type	Description	Default
image_url	str	Docker image URL to pull.	required

Source code in src/nskit/client/backends/docker.py

def pull_image(self, image_url: str) -> None:
    """Pull Docker image.

    Args:
        image_url: Docker image URL to pull.
    """
    validate_image_url(image_url)
    self._authenticate()
    subprocess.run(  # nosec B603, B607
        ["docker", "pull", image_url], check=True, capture_output=True, timeout=self.timeouts.pull
    )

nskit.client.backends.docker_local.DockerLocalBackend

Bases: RecipeBackend

Backend that discovers recipes from locally available Docker images.

Filters images by the nskit.recipe=true label. Each image must also have nskit.recipe.name set. Versions are derived from image tags.

Parameters:

Name	Type	Description	Default
entrypoint	str	Recipe entry point group name.	RECIPE_ENTRYPOINT
label_filter	str	Label to filter images (default nskit.recipe=true).	f'{LABEL_PREFIX}=true'

Source code in src/nskit/client/backends/docker_local.py

class DockerLocalBackend(RecipeBackend):
    """Backend that discovers recipes from locally available Docker images.

    Filters images by the ``nskit.recipe=true`` label. Each image must
    also have ``nskit.recipe.name`` set. Versions are derived from image
    tags.

    Args:
        entrypoint: Recipe entry point group name.
        label_filter: Label to filter images (default ``nskit.recipe=true``).
    """

    def __init__(
        self,
        entrypoint: str = RECIPE_ENTRYPOINT,
        label_filter: str = f"{LABEL_PREFIX}=true",
    ) -> None:
        self._entrypoint = entrypoint
        self._label_filter = label_filter

    @property
    def entrypoint(self) -> str:
        """Get the recipe entrypoint."""
        return self._entrypoint

    def _query_images(self) -> list[dict[str, str]]:
        """Query Docker for images matching the label filter.

        Returns:
            List of dicts with ``repository``, ``tag``, and ``name`` keys.
        """
        from nskit.client.backends.image_labels import get_recipe_name, read_local_labels

        result = subprocess.run(  # nosec B603, B607
            [
                "docker",
                "images",
                "--filter",
                f"label={self._label_filter}",
                "--format",
                "{{.Repository}}\t{{.Tag}}",
            ],
            capture_output=True,
            text=True,
            check=False,
        )
        images = []
        for line in result.stdout.strip().splitlines():
            parts = line.split("\t")
            if len(parts) >= 2 and parts[1] != "<none>":
                image_ref = f"{parts[0]}:{parts[1]}"
                labels = read_local_labels(image_ref)
                name = get_recipe_name(labels) or parts[0].rsplit("/", 1)[-1]
                images.append({"repository": parts[0], "tag": parts[1], "name": name})
        return images

    def list_recipes(self) -> list[RecipeInfo]:
        """List recipes from locally pulled Docker images.

        Returns:
            List of recipe information from local images.
        """
        images = self._query_images()
        recipes: dict[str, list[str]] = {}
        for img in images:
            recipes.setdefault(img["name"], []).append(img["tag"])
        return [RecipeInfo(name=name, versions=sorted(versions, reverse=True)) for name, versions in recipes.items()]

    def get_recipe_versions(self, recipe: str) -> list[str]:
        """Get versions for a recipe from local image tags.

        Args:
            recipe: Recipe name.

        Returns:
            Sorted list of version strings (newest first).
        """
        images = self._query_images()
        return sorted([img["tag"] for img in images if img["name"] == recipe], reverse=True)

    def fetch_recipe(self, recipe: str, version: str, dest: Path) -> Path:
        """Not used — Docker images are executed directly.

        Raises:
            NotImplementedError: Always.
        """
        raise NotImplementedError("DockerLocalBackend executes images directly")

    def get_image_url(self, recipe: str, version: str) -> str:
        """Get the full image:tag for a recipe version.

        Args:
            recipe: Recipe name.
            version: Recipe version.

        Returns:
            Full image reference string.

        Raises:
            ValueError: If no matching local image is found.
        """
        images = self._query_images()
        for img in images:
            if img["name"] == recipe and img["tag"] == version:
                return f"{img['repository']}:{img['tag']}"
        raise ValueError(f"No local image found for {recipe}:{version}")

    def pull_image(self, image_url: str) -> None:
        """No-op — images are already local."""

entrypoint property

Get the recipe entrypoint.

fetch_recipe(recipe, version, dest)

Not used — Docker images are executed directly.

Raises:

Type	Description
NotImplementedError	Always.

Source code in src/nskit/client/backends/docker_local.py

def fetch_recipe(self, recipe: str, version: str, dest: Path) -> Path:
    """Not used — Docker images are executed directly.

    Raises:
        NotImplementedError: Always.
    """
    raise NotImplementedError("DockerLocalBackend executes images directly")

get_image_url(recipe, version)

Get the full image:tag for a recipe version.

Parameters:

Name	Type	Description	Default
recipe	str	Recipe name.	required
version	str	Recipe version.	required

Returns:

Type	Description
str	Full image reference string.

Raises:

Type	Description
ValueError	If no matching local image is found.

Source code in src/nskit/client/backends/docker_local.py

def get_image_url(self, recipe: str, version: str) -> str:
    """Get the full image:tag for a recipe version.

    Args:
        recipe: Recipe name.
        version: Recipe version.

    Returns:
        Full image reference string.

    Raises:
        ValueError: If no matching local image is found.
    """
    images = self._query_images()
    for img in images:
        if img["name"] == recipe and img["tag"] == version:
            return f"{img['repository']}:{img['tag']}"
    raise ValueError(f"No local image found for {recipe}:{version}")

get_recipe_versions(recipe)

Get versions for a recipe from local image tags.

Parameters:

Name	Type	Description	Default
recipe	str	Recipe name.	required

Returns:

Type	Description
list[str]	Sorted list of version strings (newest first).

Source code in src/nskit/client/backends/docker_local.py

def get_recipe_versions(self, recipe: str) -> list[str]:
    """Get versions for a recipe from local image tags.

    Args:
        recipe: Recipe name.

    Returns:
        Sorted list of version strings (newest first).
    """
    images = self._query_images()
    return sorted([img["tag"] for img in images if img["name"] == recipe], reverse=True)

list_recipes()

List recipes from locally pulled Docker images.

Returns:

Type	Description
list[RecipeInfo]	List of recipe information from local images.

Source code in src/nskit/client/backends/docker_local.py

def list_recipes(self) -> list[RecipeInfo]:
    """List recipes from locally pulled Docker images.

    Returns:
        List of recipe information from local images.
    """
    images = self._query_images()
    recipes: dict[str, list[str]] = {}
    for img in images:
        recipes.setdefault(img["name"], []).append(img["tag"])
    return [RecipeInfo(name=name, versions=sorted(versions, reverse=True)) for name, versions in recipes.items()]

pull_image(image_url)

No-op — images are already local.

Source code in src/nskit/client/backends/docker_local.py

def pull_image(self, image_url: str) -> None:
    """No-op — images are already local."""

Models

nskit.client.models.RecipeResult

Bases: BaseModel

Result of recipe initialization.

Source code in src/nskit/client/models.py

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)

nskit.client.models.UpdateResult

Bases: BaseModel

Result of recipe update.

Source code in src/nskit/client/models.py

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.client.models.RecipeInfo

Bases: BaseModel

Information about a recipe.

Source code in src/nskit/client/models.py

class RecipeInfo(BaseModel):
    """Information about a recipe."""

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

Configuration

nskit.client.config.ConfigManager

Manages recipe configuration files.

Parameters:

Name	Type	Description	Default
project_path	Path	Root path of the project.	required
config_dir	str	Directory name for config storage (default .recipe).	'.recipe'
config_filename	str	Config file name (default config.yml).	'config.yml'

Source code in src/nskit/client/config.py

class ConfigManager:
    """Manages recipe configuration files.

    Args:
        project_path: Root path of the project.
        config_dir: Directory name for config storage (default ``.recipe``).
        config_filename: Config file name (default ``config.yml``).
    """

    def __init__(
        self,
        project_path: Path,
        config_dir: str = ".recipe",
        config_filename: str = "config.yml",
    ) -> None:
        self.project_path = project_path
        self.config_dir = config_dir
        self.config_filename = config_filename
        self.config_path = project_path / config_dir / config_filename

    def load_config(self) -> RecipeConfig:
        """Load recipe configuration from YAML.

        Returns:
            Parsed ``RecipeConfig`` instance.

        Raises:
            ProjectNotRecipeBasedError: If the config file does not exist.
            InvalidConfigError: If the YAML is invalid or cannot be parsed.
        """
        if not self.config_path.exists():
            raise ProjectNotRecipeBasedError(str(self.project_path))

        raw = self.config_path.read_text(encoding="utf-8")
        try:
            data = yaml.safe_load(raw)
        except yaml.YAMLError as exc:
            raise InvalidConfigError([str(exc)]) from exc

        if data is None:
            data = {}

        if not isinstance(data, dict):
            raise InvalidConfigError([f"Expected a mapping, got {type(data).__name__}"])

        try:
            return RecipeConfig.model_validate(data)
        except Exception as exc:
            raise InvalidConfigError([str(exc)]) from exc

    def save_config(self, config: RecipeConfig) -> None:
        """Save recipe configuration to YAML.

        Creates the config directory if it does not exist. Serialises
        datetime fields to ISO format strings.

        Args:
            config: Configuration to persist.
        """
        self.config_path.parent.mkdir(parents=True, exist_ok=True)
        data = config.model_dump(mode="json")
        content = yaml.dump(data, default_flow_style=False, sort_keys=False)
        self.config_path.write_text(content, encoding="utf-8")

    def update_config_version(self, new_version: str, recipe_name: str) -> None:
        """Update the Docker image version and timestamp in the config.

        Args:
            new_version: New version tag to set.
            recipe_name: Recipe name (used if metadata is missing).
        """
        config = self.load_config()
        if config.metadata is None:
            config.metadata = RecipeMetadata(
                recipe_name=recipe_name,
                docker_image=f"{recipe_name}:{new_version}",
            )
        else:
            # Replace version tag in docker_image URL
            current_image = config.metadata.docker_image
            config.metadata.docker_image = re.sub(r":[^/]+$", f":{new_version}", current_image)
        config.metadata.updated_at = datetime.now(tz=timezone.utc)
        self.save_config(config)

    def is_recipe_based(self) -> bool:
        """Check whether the project has a recipe configuration file.

        Returns:
            ``True`` if the config file exists, ``False`` otherwise.
        """
        return self.config_path.exists()

is_recipe_based()

Check whether the project has a recipe configuration file.

Returns:

Type	Description
bool	True if the config file exists, False otherwise.

Source code in src/nskit/client/config.py

def is_recipe_based(self) -> bool:
    """Check whether the project has a recipe configuration file.

    Returns:
        ``True`` if the config file exists, ``False`` otherwise.
    """
    return self.config_path.exists()

load_config()

Load recipe configuration from YAML.

Returns:

Type	Description
RecipeConfig	Parsed RecipeConfig instance.

Raises:

Type	Description
ProjectNotRecipeBasedError	If the config file does not exist.
InvalidConfigError	If the YAML is invalid or cannot be parsed.

Source code in src/nskit/client/config.py

def load_config(self) -> RecipeConfig:
    """Load recipe configuration from YAML.

    Returns:
        Parsed ``RecipeConfig`` instance.

    Raises:
        ProjectNotRecipeBasedError: If the config file does not exist.
        InvalidConfigError: If the YAML is invalid or cannot be parsed.
    """
    if not self.config_path.exists():
        raise ProjectNotRecipeBasedError(str(self.project_path))

    raw = self.config_path.read_text(encoding="utf-8")
    try:
        data = yaml.safe_load(raw)
    except yaml.YAMLError as exc:
        raise InvalidConfigError([str(exc)]) from exc

    if data is None:
        data = {}

    if not isinstance(data, dict):
        raise InvalidConfigError([f"Expected a mapping, got {type(data).__name__}"])

    try:
        return RecipeConfig.model_validate(data)
    except Exception as exc:
        raise InvalidConfigError([str(exc)]) from exc

save_config(config)

Save recipe configuration to YAML.

Creates the config directory if it does not exist. Serialises datetime fields to ISO format strings.

Parameters:

Name	Type	Description	Default
config	RecipeConfig	Configuration to persist.	required

Source code in src/nskit/client/config.py

def save_config(self, config: RecipeConfig) -> None:
    """Save recipe configuration to YAML.

    Creates the config directory if it does not exist. Serialises
    datetime fields to ISO format strings.

    Args:
        config: Configuration to persist.
    """
    self.config_path.parent.mkdir(parents=True, exist_ok=True)
    data = config.model_dump(mode="json")
    content = yaml.dump(data, default_flow_style=False, sort_keys=False)
    self.config_path.write_text(content, encoding="utf-8")

update_config_version(new_version, recipe_name)

Update the Docker image version and timestamp in the config.

Parameters:

Name	Type	Description	Default
new_version	str	New version tag to set.	required
recipe_name	str	Recipe name (used if metadata is missing).	required

Source code in src/nskit/client/config.py

def update_config_version(self, new_version: str, recipe_name: str) -> None:
    """Update the Docker image version and timestamp in the config.

    Args:
        new_version: New version tag to set.
        recipe_name: Recipe name (used if metadata is missing).
    """
    config = self.load_config()
    if config.metadata is None:
        config.metadata = RecipeMetadata(
            recipe_name=recipe_name,
            docker_image=f"{recipe_name}:{new_version}",
        )
    else:
        # Replace version tag in docker_image URL
        current_image = config.metadata.docker_image
        config.metadata.docker_image = re.sub(r":[^/]+$", f":{new_version}", current_image)
    config.metadata.updated_at = datetime.now(tz=timezone.utc)
    self.save_config(config)

nskit.client.config.RecipeConfig

Bases: BaseModel

Recipe configuration persisted in a project.

Parameters:

Name	Type	Description	Default
input		Original input parameters used for generation.	required
rendered		Rendered/resolved parameter values.	required
metadata		Recipe metadata (name, image, timestamps).	required

Source code in src/nskit/client/config.py

class RecipeConfig(BaseModel):
    """Recipe configuration persisted in a project.

    Args:
        input: Original input parameters used for generation.
        rendered: Rendered/resolved parameter values.
        metadata: Recipe metadata (name, image, timestamps).
    """

    input: dict[str, Any] = Field(default_factory=dict)
    rendered: dict[str, Any] = Field(default_factory=dict)
    metadata: RecipeMetadata | None = None

Exceptions

nskit.client.exceptions

Domain-specific exceptions for nskit client operations.

ConfigNotFoundError

Bases: Exception

Raised when the recipe configuration file does not exist.

Parameters:

Name	Type	Description	Default
config_path	str	Expected path to the configuration file.	required

Source code in src/nskit/client/exceptions.py

class ConfigNotFoundError(Exception):
    """Raised when the recipe configuration file does not exist.

    Args:
        config_path: Expected path to the configuration file.
    """

    def __init__(self, config_path: str) -> None:
        self.config_path = config_path
        super().__init__(str(self))

    def __str__(self) -> str:
        """Format the error message with the expected config path."""
        return f"Recipe configuration file not found at '{self.config_path}'"

__str__()

Format the error message with the expected config path.

Source code in src/nskit/client/exceptions.py

def __str__(self) -> str:
    """Format the error message with the expected config path."""
    return f"Recipe configuration file not found at '{self.config_path}'"

FileSystemError

Bases: Exception

Raised when a file system operation fails.

Parameters:

Name	Type	Description	Default
operation	str	Name of the operation that failed.	required
path	str	Path of the file involved.	required
reason	str	Description of why the operation failed.	required

Source code in src/nskit/client/exceptions.py

class FileSystemError(Exception):
    """Raised when a file system operation fails.

    Args:
        operation: Name of the operation that failed.
        path: Path of the file involved.
        reason: Description of why the operation failed.
    """

    def __init__(self, operation: str, path: str, reason: str) -> None:
        self.operation = operation
        self.path = path
        self.reason = reason
        super().__init__(str(self))

    def __str__(self) -> str:
        """Format the error message with operation, path, and reason."""
        return f"File system error during '{self.operation}' on '{self.path}': {self.reason}"

__str__()

Format the error message with operation, path, and reason.

Source code in src/nskit/client/exceptions.py

def __str__(self) -> str:
    """Format the error message with operation, path, and reason."""
    return f"File system error during '{self.operation}' on '{self.path}': {self.reason}"

GitStatusError

Bases: Exception

Raised when the Git repository is not in a ready state.

Parameters:

Name	Type	Description	Default
reason	str	Description of why the repository is not ready.	required

Source code in src/nskit/client/exceptions.py

class GitStatusError(Exception):
    """Raised when the Git repository is not in a ready state.

    Args:
        reason: Description of why the repository is not ready.
    """

    def __init__(self, reason: str) -> None:
        self.reason = reason
        super().__init__(str(self))

    def __str__(self) -> str:
        """Format the error message with the reason."""
        return f"Git repository is not ready: {self.reason}"

__str__()

Format the error message with the reason.

Source code in src/nskit/client/exceptions.py

def __str__(self) -> str:
    """Format the error message with the reason."""
    return f"Git repository is not ready: {self.reason}"

InitError

Bases: Exception

Raised when recipe initialisation fails.

Parameters:

Name	Type	Description	Default
message	str	Description of the initialisation failure.	required
details	str | None	Optional additional details about the failure.	None

Source code in src/nskit/client/exceptions.py

class InitError(Exception):
    """Raised when recipe initialisation fails.

    Args:
        message: Description of the initialisation failure.
        details: Optional additional details about the failure.
    """

    def __init__(self, message: str, details: str | None = None) -> None:
        self.message = message
        self.details = details
        super().__init__(str(self))

    def __str__(self) -> str:
        """Format the error message with optional details."""
        if self.details:
            return f"{self.message}\nDetails: {self.details}"
        return self.message

__str__()

Format the error message with optional details.

Source code in src/nskit/client/exceptions.py

def __str__(self) -> str:
    """Format the error message with optional details."""
    if self.details:
        return f"{self.message}\nDetails: {self.details}"
    return self.message

InvalidConfigError

Bases: Exception

Raised when a recipe configuration file contains invalid content.

Parameters:

Name	Type	Description	Default
errors	list[str]	List of validation error descriptions.	required

Source code in src/nskit/client/exceptions.py

class InvalidConfigError(Exception):
    """Raised when a recipe configuration file contains invalid content.

    Args:
        errors: List of validation error descriptions.
    """

    def __init__(self, errors: list[str]) -> None:
        self.errors = errors
        super().__init__(str(self))

    def __str__(self) -> str:
        """Format the error message with a bulleted list of errors."""
        bullet_list = "\n".join(f"  - {error}" for error in self.errors)
        return f"Invalid recipe configuration:\n{bullet_list}"

__str__()

Format the error message with a bulleted list of errors.

Source code in src/nskit/client/exceptions.py

def __str__(self) -> str:
    """Format the error message with a bulleted list of errors."""
    bullet_list = "\n".join(f"  - {error}" for error in self.errors)
    return f"Invalid recipe configuration:\n{bullet_list}"

ProjectNotRecipeBasedError

Bases: Exception

Raised when a project does not have a recipe configuration file.

Parameters:

Name	Type	Description	Default
project_path	str	Path to the project directory.	required

Source code in src/nskit/client/exceptions.py

class ProjectNotRecipeBasedError(Exception):
    """Raised when a project does not have a recipe configuration file.

    Args:
        project_path: Path to the project directory.
    """

    def __init__(self, project_path: str) -> None:
        self.project_path = project_path
        super().__init__(str(self))

    def __str__(self) -> str:
        """Format the error message with project path and guidance."""
        return (
            f"Project at '{self.project_path}' is not recipe-based. "
            f"Expected a recipe configuration file in the project directory."
        )

__str__()

Format the error message with project path and guidance.

Source code in src/nskit/client/exceptions.py

def __str__(self) -> str:
    """Format the error message with project path and guidance."""
    return (
        f"Project at '{self.project_path}' is not recipe-based. "
        f"Expected a recipe configuration file in the project directory."
    )

UpdateError

Bases: Exception

Raised when a recipe update operation fails.

Parameters:

Name	Type	Description	Default
message	str	Description of the update failure.	required
details	str | None	Optional additional details about the failure.	None

Source code in src/nskit/client/exceptions.py

class UpdateError(Exception):
    """Raised when a recipe update operation fails.

    Args:
        message: Description of the update failure.
        details: Optional additional details about the failure.
    """

    def __init__(self, message: str, details: str | None = None) -> None:
        self.message = message
        self.details = details
        super().__init__(str(self))

    def __str__(self) -> str:
        """Format the error message with optional details."""
        if self.details:
            return f"{self.message}\nDetails: {self.details}"
        return self.message

__str__()

Format the error message with optional details.

Source code in src/nskit/client/exceptions.py

def __str__(self) -> str:
    """Format the error message with optional details."""
    if self.details:
        return f"{self.message}\nDetails: {self.details}"
    return self.message