Skip to content
Back to skills

2522 Pipeline Spec A1346c91

ASecurity

`PipelineSpec` provides a fluent API for constructing middleware pipelines that can be compiled into middleware instances or exported as configuration files.

  • 9 stars
  • 0 votes
  • 0 copies
  • 0 views
  • Added October 11, 2026
devopspythonnodeapi

Works with

  • api

Security analysis

A100/100

Scanned October 11, 2026

npx -y skills add tools-only/X-Skills --skill 2522-pipeline-spec_a1346c91 --agent claude-code

Installs into .claude/skills of the current project.

Are you the author of 2522 Pipeline Spec A1346c91?

Add the live security badge to your README. It updates with every re-scan.

Security grade badge for 2522 Pipeline Spec A1346c91
[![Security: A — Skills Directory](https://www.skillsdirectory.com/api/skills/tools-only-2522-pipeline-spec-a1346c91/badge)](https://www.skillsdirectory.com/skills/tools-only-2522-pipeline-spec-a1346c91)

More formats (shields.io, HTML) on the badges page. Keep it an A: scan every change in CI with Pro.

SKILL.md
# Pipeline Spec

`PipelineSpec` provides a fluent API for constructing middleware pipelines that
can be compiled into middleware instances or exported as configuration files.

## Quick Example

```python
from pydantic_ai_middleware import PipelineSpec
from pydantic_ai_middleware.builder import MiddlewarePipelineCompiler

# Build a pipeline spec with fluent API
spec = (
    PipelineSpec()
    .add_type("logging", {"level": "DEBUG"})
    .add_type("rate_limit", {"max_requests": 100})
    .add_when(
        predicate="is_admin",
        then=[{"type": "admin_audit"}],
        else_=[{"type": "user_audit"}],
    )
)

# Export to YAML file
spec.save("middleware-pipeline.yaml")

# Or compile directly to middleware instances
compiler = MiddlewarePipelineCompiler(registry)
middleware_list = spec.compile(compiler)
```

## Supported Node Types

- **type**: A single middleware by registered name
- **chain**: Sequential execution of multiple nodes
- **parallel**: Concurrent execution with result aggregation
- **when**: Conditional branching based on predicates

## Export as JSON or YAML

```python
from pydantic_ai_middleware.pipeline_spec import PipelineSpec

PipelineSpec().add_type("logging").save("pipeline.json")
PipelineSpec().add_type("logging").save("pipeline.yaml")

# Or get as string
yaml_str = PipelineSpec().add_type("logging").dump("yaml")
json_str = PipelineSpec().add_type("logging").dump("json")
```

## Next Steps

- [Config Loading](config-loading.md) - Load pipelines from config files
- [API Reference](../api/pipeline_spec.md) - PipelineSpec API

Attribution

Is this your skill, or is something wrong with this listing? Request removal or report an issue. Author removals are honored within 72 hours.

Comments

Loading comments…