Python API Reference

The aidin package provides programmatic Python interfaces for defining components, assembling pipelines, compiling artifacts, and executing runs on Aidin Pipelines.

import aidin

# Or import primary symbols directly:
from aidin import (
    ProjectClient,
    compile,
    compile_pipeline,
    component,
    pipeline,
)

Compilation

`aidin.compile`

Compiles a pipeline function into a KFP v2 YAML pipeline specification package, optionally building a Docker container image. aidin.compile_pipeline is an exact alias.

def compile(
    pipeline_func: Any,
    package_path: str,
    pipeline_name: Optional[str] = None,
    pipeline_parameters: Optional[Dict[str, Any]] = None,
    type_check: bool = True,
    build: bool = False,
    build_root_directory: Optional[str] = None,
    build_registry_address: Optional[str] = None,
) -> Optional[str]:

Parameters

Parameter	Type	Default	Description
`pipeline_func`	`Callable`	(Required)	The Python function decorated with `@pipeline`.
`package_path`	`str`	(Required)	Destination file path for the output YAML specification.
`pipeline_name`	`Optional[str]`	`None`	Custom name for the pipeline. Defaults to the sanitized function name.
`pipeline_parameters`	`Optional[Dict[str, Any]]`	`None`	Default parameters to embed into the compiled pipeline definition.
`type_check`	`bool`	`True`	Whether to perform static type validation across connected task inputs/outputs.
`build`	`bool`	`False`	When `True`, builds a container image containing project code and rewrites YAML image paths.
`build_root_directory`	`Optional[str]`	`None`	Path to the root directory of the project. Required when `build=True`.
`build_registry_address`	`Optional[str]`	`None`	OCI registry destination for pushed images. Defaults to `$AIDIN_REGISTRY_ADDRESS` or `"harbor.aidin-ai.com/aidin-platform"`.

Returns

Optional[str]: The built image tag (e.g. harbor.aidin-ai.com/aidin-platform/my-pipeline:20261109123456) if build=True, otherwise None.

Raises

ImportError: If kfp is not installed.
ValueError: If build=True but build_root_directory is omitted.
FileNotFoundError: If the output YAML file cannot be created or read for rewriting.

Example

from aidin import compile
from my_project.pipelines import training_pipeline

# Compile basic pipeline:
compile(
    pipeline_func=training_pipeline,
    package_path="dist/training_pipeline.yaml",
)

# Compile with unified container build:
image = compile(
    pipeline_func=training_pipeline,
    package_path="dist/training_pipeline.yaml",
    build=True,
    build_root_directory=".",
    build_registry_address="harbor.aidin-ai.com/aidin-platform",
)
print(f"Compiled pipeline with image: {image}")

DSL and Component Decorators

`aidin.component`

Defines a lightweight container component from a Python function, injecting configurable package repositories and default images.

def component(
    base_image: Optional[str] = None,
    packages_to_install: List[str] = [],
) -> Callable:

Parameters

Parameter	Type	Default	Description
`base_image`	`Optional[str]`	`None`	Container image in which to run the component. If omitted, uses `$COMPS_BASE_IMAGE` or `python:3.11`.
`packages_to_install`	`List[str]`	`[]`	Python package names or requirement specifiers to install at component runtime.

Aidin Enhancements

Components decorated with aidin.component support custom package indexes configured via:

export AIDIN_PIP_INDEX_URL="https://hamgit.ir/api/v4/projects/23708/packages/pypi/simple"

If not specified, dependencies are installed using standard PyPI/environment index settings.

Re-exported DSL Types in `aidin.dsl`

In addition to component and pipeline, aidin.dsl re-exports all standard Kubeflow Pipelines v2 types:

Data Artifacts: Input, Output, Dataset, Model, Artifact, HTML, Markdown
Metrics: Metrics, ClassificationMetrics, SlicedClassificationMetrics
Control Flow: Condition, If, Elif, Else, ParallelFor, Collected, ExitHandler

Example

from aidin.dsl import component, Input, Output, Dataset, Model, Metrics

@component(packages_to_install=["pandas", "scikit-learn"])
def train_model(
    train_data: Input[Dataset],
    model_artifact: Output[Model],
    metrics: Output[Metrics],
    n_estimators: int = 100,
) -> float:
    import pandas as pd
    from sklearn.ensemble import RandomForestClassifier
    import joblib

    df = pd.read_csv(train_data.path)
    X, y = df.drop(columns=["target"]), df["target"]

    clf = RandomForestClassifier(n_estimators=n_estimators)
    clf.fit(X, y)

    score = float(clf.score(X, y))
    metrics.log_metric("accuracy", score)

    joblib.dump(clf, model_artifact.path)
    return score

`aidin.pipeline`

Decorates a Python function that orchestrates components into a directed acyclic graph (DAG).

def pipeline(
    func: Optional[Callable] = None,
    *,
    name: Optional[str] = None,
    description: Optional[str] = None,
) -> Callable:

Parameters

Parameter	Type	Default	Description
`func`	`Optional[Callable]`	`None`	The pipeline orchestration function (when used as `@pipeline` without arguments).
`name`	`Optional[str]`	`None`	Display name of the pipeline. Defaults to a sanitized version of the function name.
`description`	`Optional[str]`	`None`	Human-readable description of the pipeline.

Example

from aidin.dsl import pipeline
from my_components import load_dataset, train_model, evaluate_model

@pipeline(
    name="customer-churn-prediction",
    description="Loads data, trains a random forest model, and evaluates test metrics."
)
def churn_pipeline(dataset_url: str, n_estimators: int = 100):
    dataset_task = load_dataset(url=dataset_url)
    train_task = train_model(
        train_data=dataset_task.outputs["dataset"],
        n_estimators=n_estimators,
    )
    evaluate_model(
        model=train_task.outputs["model_artifact"],
        test_data=dataset_task.outputs["dataset"],
    )

Client Integration

`aidin.ProjectClient`

A subclass of kfp.Client configured for Aidin clusters using short-lived ServiceAccount tokens obtained via the Aidin Token Exchanger.

class ProjectClient(kfp.Client):
    def __init__(
        self,
        authentik_token: Optional[str] = None,
        project_name: Optional[str] = None,
        exchanger_url: Optional[str] = None,
        pipeline_server_url: Optional[str] = None,
        verify_ssl: bool = True,
        force_refresh: bool = False,
        aidin_user_dir: Optional[str] = None,
        *,
        timeout: float = 30.0,
        **kwargs: Any,
    )

Parameters

Parameter	Type	Default	Description
`authentik_token`	`Optional[str]`	`None`	Upstream Authentik bearer token. Can be `None` if a valid cached token exists. When passed as a single positional argument, it is interpreted as `project_name`.
`project_name`	`Optional[str]`	`None`	Aidin project name (required either as positional argument or keyword argument).
`exchanger_url`	`Optional[str]`	`None`	Token Exchanger URL. Defaults to `https://token-exchanger.<AIDIN_BASE_URL>`.
`pipeline_server_url`	`Optional[str]`	`None`	Aidin Pipelines API URL. Defaults to `https://pipeline-server.<AIDIN_BASE_URL>`.
`verify_ssl`	`bool`	`True`	Whether to verify TLS certificates. When `False`, automatically configures `KFP_SKIP_TLS_VERIFY="true"`.
`force_refresh`	`bool`	`False`	If `True`, requests a new service token even if a valid cached token exists.
`aidin_user_dir`	`Optional[str]`	`None`	Base directory for credentials cache. Defaults to `~/.aidin/`.
`timeout`	`float`	`30.0`	HTTP request timeout in seconds for exchange requests.
`**kwargs`	`Any`		Additional keyword arguments forwarded directly to `kfp.Client.__init__`.

Custom Instance Attributes

client.aidin_project_name: str — Name of the connected project profile.
client.aidin_namespace: str — Kubernetes namespace associated with the project.
client.aidin_expires_at: Optional[str] — ISO-8601 UTC timestamp of service token expiration.

Inherited `kfp.Client` Methods

ProjectClient exposes the complete kfp.Client feature set:

upload_pipeline(pipeline_package_path, pipeline_name=None, description=None)
create_run_from_pipeline_func(pipeline_func, arguments=None, experiment_name=None, ...)
create_run_from_pipeline_package(pipeline_file, arguments=None, experiment_name=None, ...)
list_pipelines(page_token=None, page_size=10, sort_by=None)
get_experiment(experiment_id=None, experiment_name=None, namespace=None)
create_experiment(name, description=None, namespace=None)
wait_for_run_completion(run_id, timeout=None)

Example: Upload and Run Execution

import os
from aidin import ProjectClient
from my_pipeline import churn_pipeline

# Initialize client using active credentials:
client = ProjectClient(
    authentik_token=os.getenv("AUTHENTIK_TOKEN"),
    project_name="fraud-detection",
)

print(f"Connected to namespace: {client.aidin_namespace}")
print(f"Token expires at: {client.aidin_expires_at}")

# Submit run directly from a Python function:
run = client.create_run_from_pipeline_func(
    pipeline_func=churn_pipeline,
    arguments={"dataset_url": "https://data.example.com/churn.csv", "n_estimators": 200},
    experiment_name="churn-experiments",
)
print(f"Pipeline run submitted: {run.run_id}")

Next Steps

Review the CLI Reference for terminal commands.
Read Authentication & Token Lifecycle for caching and security details.

Python API Reference

Compilation

aidin.compile

Parameters

Returns

Raises

Example

DSL and Component Decorators

aidin.component

Parameters

Aidin Enhancements

Re-exported DSL Types in aidin.dsl

Example

aidin.pipeline

Parameters

Example

Client Integration

aidin.ProjectClient

Parameters

Custom Instance Attributes

Inherited kfp.Client Methods

Example: Upload and Run Execution

Next Steps

`aidin.compile`

`aidin.component`

Re-exported DSL Types in `aidin.dsl`

`aidin.pipeline`

`aidin.ProjectClient`

Inherited `kfp.Client` Methods