#rlsbl.pipelines.base

#rlsbl.pipelines.base

Base classes for release pipelines, including no-op defaults for optional methods and token/credential authentication mixins.

#BasePipeline

Concrete base providing no-op defaults for optional Pipeline methods.

Subclasses override specific methods to implement their publish mechanism.

#__init__

def __init__(self, name: str, pipeline_type: str, local: bool, config: dict)

Initialize with pipeline identity, local-publish flag, and config dict.

#probe_before_publish

def probe_before_publish(self, dir_path: str, version: str, ctx) -> bool

Probe the linked target's registry before publishing.

Returns True if publishing should proceed, False if already published (skip with log). Requires the pipeline to have a target attribute linking it to a target name.

#is_already_published_error

def is_already_published_error(exc: Exception) -> bool

Check if an exception represents an "already published" error.

Inspects the exception message, stdout, and stderr for known registry error signatures. When True, the caller should treat the error as success (idempotent).

#publish

def publish(self, dir_path: str, version: str, ctx) -> None

Publish the package. No-op by default; subclasses override.

#build_assets

def build_assets(self, dir_path: str, version: str, dist_dir: str, ctx) -> list[str]

Build release assets into dist_dir. Returns paths; empty by default.

#template_dir

def template_dir(self) -> str | None

Return the directory containing CI templates, or None if none.

#template_mappings

def template_mappings(self, ctx) -> list[dict[str, str]]

Return template-to-target mappings for CI scaffold. Empty by default.

#_linked_target_subdir

def _linked_target_subdir(self, ctx) -> str

Resolve the directory of this pipeline's linked target.

Reads ctx.config["targets"] for an entry whose name matches this pipeline's target link and returns its path ("." for a root target, or a subdirectory like "packaging/npm"). Returns "." when the target is unknown or ctx is unavailable (e.g. unit tests passing ctx=None).

#build_custom_assets

def build_custom_assets(self, dist_dir: str) -> list[str]

Build custom assets defined in the pipeline config.

Reads self.config["custom_assets"] (list of dicts with name and build keys), runs each build command via shell subprocess, verifies the output file exists, and checks its size against self.config["max_asset_size_mb"].

Returns a list of absolute paths to the built asset files. Returns an empty list if no custom_assets are configured.

#required_env_vars

def required_env_vars(self) -> list[str]

Return env var names required for local publish. Empty by default.

#TokenPipeline

Base for pipelines that authenticate via a single token env var.

Subclasses set _default_token_var and implement _publish_command.

#__init__

def __init__(self, name: str, pipeline_type: str, local: bool, config: dict)

Initialize and resolve the token env var from config or class default.

#publish

def publish(self, dir_path: str, version: str, ctx) -> None

Resolve the token and delegate to _publish_command.

#required_env_vars

def required_env_vars(self) -> list[str]

Return the token var when local publish is enabled.

#CredentialPipeline

Base for pipelines that authenticate via username/password env vars.

Subclasses set _default_username_var and _default_password_var and implement _publish_command.

#__init__

def __init__(self, name: str, pipeline_type: str, local: bool, config: dict)

Initialize and resolve username/password env vars from config or defaults.

#publish

def publish(self, dir_path: str, version: str, ctx) -> None

Resolve credentials and delegate to _publish_command.

#required_env_vars

def required_env_vars(self) -> list[str]

Return username and password vars when local publish is enabled.