Bclearer Pipeline Engineer

bclearer Pipeline Engineer

Role

You are a bclearer pipeline engineer. You extend the ob-engineer role with specialised knowledge of bclearer pipeline code conventions, interop service usage, and orchestration wiring. bclearer is an OB-specific framework — all BORO Quick Style Guide conventions from ob-engineer apply in full, plus the bclearer-specific additions in this file.

Read `skills/ob-engineer/SKILL.md` first (which itself extends `skills/python-data-engineer/SKILL.md` and `skills/data-engineer/SKILL.md`) and follow all of it. This file contains only the additions and overrides that apply specifically to bclearer pipeline work.

Note: references/bclearer-code-style.md overrides the general Python formatting conventions — use bclearer conventions (backslash continuations, named kwargs) throughout.

Additional Knowledge

Beyond the base data-engineer references, you draw on:

The base data-engineer references (clean-coding-index.md, testing-index.md) remain fully in scope, but references/bclearer-code-style.md takes precedence for formatting and naming where it specifies a stricter rule.

Sub-skill Delegation

In addition to the clean coding sub-skills inherited from data-engineer, you delegate:

---

bclearer-Specific Additions to Implement Mode

Apply these additions on top of the base Implement Mode workflow.

Construction Order for Pipelines (Step 3 addition)

Follow the approved pipeline topology from the architecture design. Within each stage:

1. Common knowledge (enums, types, constants)
2. Domain objects / BIE components (delegate to bie-data-engineer)
3. Stage adapters (ingest and load/export)
4. Stage processors/services (transform/enrich)
5. Stage orchestrators (wires adapters + services)
6. Pipeline runner / entry point (wires all stages)

Code Layout Convention

Each pipeline lives under a dedicated package. Recommended structure:

[pipeline_name]/
├── common_knowledge/       # pipeline-level enums, types, constants
├── bie/                    # BIE domain objects (if applicable; bie-data-engineer's output)
├── adapters/
│   ├── ingest/             # Stage 1 adapters
│   └── export/             # Stage 4 adapters
├── services/               # Stage 3 processing/transform logic
├── orchestrators/          # Stage orchestrators
└── runners/                # Entry point(s)

See references/pipeline-implementation.md for file-level conventions within each folder.

Additional Verification (Step 5 addition)

Beyond pytest/mypy/ruff, verify:

• [ ] Each stage is independently testable (no direct cross-stage imports)
• [ ] Interop services appear only in adapters/ — not in services/ or orchestrators/
• [ ] BIE factories only in bie/ — not in services/ or adapters/
• [ ] Universe is created in the runner, not in a stage
• [ ] No module-level or global mutable state

---

bclearer-Specific Additions to Review Mode

When reviewing bclearer pipeline code, add to the standard review checklist:

bclearer Code Style

This reference mirrors bie-data-engineer/references/code-style.md and extends it

with any pipeline-specific additions. Where this file specifies a rule, it takes

precedence over the general clean coding standards from data-engineer/references/clean-coding-index.md.

---

Source of Truth

The canonical bclearer code style is defined in:

skills/bie-data-engineer/references/code-style.md

Read that file for the full style guide. Key rules summarised below for quick reference.

---

Quick Reference

Line Continuations — use backslash

result = \
    some_function()
return \
    result

Named Keyword Arguments — always use after first positional

BieIdCreationFacade.create_bie_id_for_single_object(
    input_object=input_object,
    bie_domain_type=bie_domain_type)

Verbose Naming — no abbreviations

# Correct
transactions_dataframe
account_bie_id
excel_workbook_ingest_adapter

# Avoid
df
acct_id
adapter

Class Inheritance — parent on indented line

class SomePipelineOrchestrators(
        BasePipelineOrchestrators):

Function Signatures — parameters each on own line

def create_pipeline_universe(
        source_path: str,
        target_path: str) \
        -> PipelineUniverse:

Imports — full paths, backslash continuation for long imports

from bclearer_orchestration_services.identification_services.b_identity_ecosystem.objects.bie_ids import \
    BieIds

---

Pipeline-Specific Naming Additions

To be confirmed as conventions solidify.

BIE Integration Reference

This reference defines when and how the bclearer-pipeline-engineer delegates to bie-component-ontologist and bie-data-engineer for BIE domain work within a pipeline.

---

When to Delegate

Rule of thumb: If the task involves BieDomainObjects, CommonIdentityVector, three-tier creator functions, or BIE factory functions — delegate to bie-data-engineer. Everything else (adapters, services, orchestrators, runner wiring) stays in scope.

---

Boundary Within a Pipeline

Pipeline package
│
├── bie/                   ← bie-data-engineer owns this folder
│   ├── common_knowledge/  (domain types enum)
│   ├── bie_id_creators/   (identity vectors, creator functions)
│   └── objects/           (domain object classes, factories, universe)
│
├── adapters/              ← bclearer-pipeline-engineer owns these
├── services/              ← bclearer-pipeline-engineer owns these
├── orchestrators/         ← bclearer-pipeline-engineer owns these
└── runners/               ← bclearer-pipeline-engineer owns these

---

Handoff Protocol

When handing off to bie-data-engineer

Provide: 1. The approved BIE component ontology (4 deliverables — from bie-component-ontologist) 2. The target package path (bie/ folder within the pipeline package) 3. Any existing code in that folder to avoid conflicts

When receiving output from bie-data-engineer

After bie-data-engineer completes: 1. Read the factory function signatures (they define what the pipeline's Identify stage calls) 2. Read the Universe class (it defines what registries are available to other stages) 3. Wire these into the Identify orchestrator and pipeline universe

---

Identity Flow Through Pipeline Stages

Stage 1 (Ingest)
  └── produces: raw data (no BIE IDs)

Stage 2 (Identify) ← bie-data-engineer's factories are called here
  └── produces: domain objects with bie_ids in Universe registries

Stage 3 (Transform)
  └── receives: domain objects looked up from Universe by bie_id
  └── produces: enriched domain objects (same bie_ids, new relations)

Stage 4 (Export)
  └── receives: domain objects from Universe
  └── writes: target format (bie_id may be exported as stable key)

The bie_id is the stable key that flows through the pipeline — it is created once in Stage 2 and referenced in all subsequent stages.

Pipeline Implementation Reference

Grounded in the bclearer PDK templates at ol_bclearer_pdk/libraries/core/bclearer_core/pipeline_builder/templates/ and the bclearer PDK source at ol_bclearer_pdk/libraries/.

---

Canonical Reference

The canonical bclearer pipeline implementation is the File System Snapshot service:

bclearer_interop_services.file_system_service.bie_file_system_domain

Read this when implementing a pipeline that requires BIE identity construction — it demonstrates stage separation, BIE factory construction order, and Universe wiring.

---

Implementation Order

Follow this order within any pipeline implementation. Do not implement later layers before earlier ones:

1. Common knowledge           — enums, types, constants for the pipeline
2. BIE domain objects         — delegate to bie-data-engineer (if pipeline uses BIE identity)
3. Stage B-units              — implement each stage in order: 1c → 2l → 3e → 4a → 5r
4. Stage orchestrators        — wire B-units into stages
5. Thin slice orchestrator    — wire stages into the thin slice
6. Pipeline orchestrator      — wire thin slices into the pipeline
7. Pipeline runner            — wrap the pipeline with @run_and_log_function()
8. Domain pipelines runner    — aggregate all pipeline runners for the domain
9. Application entry point    — wrap the domain runner with run_b_application()

---

File Naming Conventions

Apply bclearer BORO naming (see references/bclearer-code-style.md):

---

Class Naming Conventions

---

Bare B-Unit Template

# {stage_prefix}{letter}_{description}_b_units.py

from bclearer_core.configurations.datastructure.logging_inspection_level_b_enums import (
    LoggingInspectionLevelBEnums,
)
from bclearer_core.objects.b_units import BUnits
from bclearer_orchestration_services.reporting_service.reporters.inspection_message_logger import (
    log_inspection_message,
)


class CaReadCsvBUnits(BUnits):

    def __init__(self):
        pass

    def run(
            self) \
            -> None:
        log_inspection_message(
            message='Running bUnit: {}'.format(
                self.__class__.__name__),
            logging_inspection_level_b_enum=\
                LoggingInspectionLevelBEnums.INFO)
        self.b_unit_process_function()

    def b_unit_process_function(
            self) \
            -> None:
        # implement here
        pass

---

Object-Passing B-Unit Template

# {stage_prefix}{letter}_{description}_b_units.py

from bclearer_core.objects.b_units import (
    BUnitsWithObjectPassingAndReturning,
)


class EaBuildTransactionsBUnits(
        BUnitsWithObjectPassingAndReturning):

    def __init__(
            self,
            input_object=None):
        super().__init__(
            input_object=input_object)

    def _b_unit_process_function(
            self) \
            -> None:
        # self.input_object is the Universe (or other typed object)
        # read from registries, build domain objects, write back
        pass

---

Stage Orchestrator Template

# {pipeline_name}_{stage_name}_orchestrator.py

from bclearer_orchestration_services.reporting_service.wrappers.run_and_log_function_wrapper_latest import (
    run_and_log_function,
)
from bclearer_pipelines.{domain_name}.b_source.common.operations.b_units.b_unit_creator_and_runner import (
    create_and_run_b_unit,
)
from bclearer_pipelines.{domain_name}.b_source.{pipeline_name}.objects.b_units.{pipeline_name}_1c_collect.ca_read_source_b_units import (
    CaReadSourceBUnits,
)


@run_and_log_function()
def orchestrate_{pipeline_name}_1c_collect() \
        -> None:
    __run_contained_bie_pipeline_components()


def __run_contained_bie_pipeline_components() \
        -> None:
    create_and_run_b_unit(
        b_unit_type=CaReadSourceBUnits)

For object-passing B-units, pass the Universe:

def __run_contained_bie_pipeline_components() \
        -> None:
    universe = \
        AccountsPipelineUniverses()
    create_and_run_b_unit(
        b_unit_type=CaReadSourceBUnits,
        input_object=universe)
    create_and_run_b_unit(
        b_unit_type=CbValidateInputBUnits,
        input_object=universe)

---

Thin Slice Orchestrator Template

# {thin_slice_name}_orchestrator.py

from bclearer_pipelines.{domain_name}.b_source.{pipeline_name}.orchestrators.stages.{pipeline_name}_1c_collect_orchestrator import (
    orchestrate_{pipeline_name}_1c_collect,
)
from bclearer_pipelines.{domain_name}.b_source.{pipeline_name}.orchestrators.stages.{pipeline_name}_3e_evolve_orchestrator import (
    orchestrate_{pipeline_name}_3e_evolve,
)
from bclearer_pipelines.{domain_name}.b_source.{pipeline_name}.orchestrators.stages.{pipeline_name}_5r_reuse_orchestrator import (
    orchestrate_{pipeline_name}_5r_reuse,
)


def orchestrate_{thin_slice_name}():
    __run_contained_bie_pipeline_components()


def __run_contained_bie_pipeline_components() \
        -> None:
    orchestrate_{pipeline_name}_1c_collect()
    orchestrate_{pipeline_name}_3e_evolve()
    orchestrate_{pipeline_name}_5r_reuse()

---

Pipeline Orchestrator Template

# {pipeline_name}_orchestrator.py

from bclearer_pipelines.{domain_name}.b_source.{pipeline_name}.orchestrators.thin_slices.{thin_slice_name}_orchestrator import (
    orchestrate_{thin_slice_name},
)


def orchestrate_{pipeline_name}():
    __run_contained_bie_pipeline_components()


def __run_contained_bie_pipeline_components() \
        -> None:
    orchestrate_{thin_slice_name}()

---

Pipeline Runner Template

# {pipeline_name}_runner.py

from bclearer_orchestration_services.reporting_service.wrappers.run_and_log_function_wrapper_latest import (
    run_and_log_function,
)
from bclearer_pipelines.{domain_name}.b_source.{pipeline_name}.orchestrators.pipeline.{pipeline_name}_orchestrator import (
    orchestrate_{pipeline_name},
)


@run_and_log_function()
def run_{pipeline_name}() \
        -> None:
    orchestrate_{pipeline_name}()

---

Application Entry Point Template

# {domain_name}_b_clearer_pipeline_b_application_runner.py

from bclearer_orchestration_services.b_app_runner_service.b_application_runner import (
    run_b_application,
)
from bclearer_pipelines.{domain_name}.b_source.app_runners.runners.{domain_name}_b_clearer_pipelines_runner import (
    run_{domain_name}_b_clearer_pipelines,
)


def run_{domain_name}_b_clearer_pipeline_b_application() \
        -> None:
    run_b_application(
        app_startup_method=run_{domain_name}_b_clearer_pipelines)

---

Universe Template

Universe holds all registries and configuration for a pipeline run:

# {pipeline_name}_universes.py

from bclearer_core.objects.universes.universes import Universes
from bclearer_pipelines.{domain_name}.b_source.{pipeline_name}.objects.enums.registry_names_enums import (
    RegistryNamesEnums,
)


class AccountsPipelineUniverses(Universes):

    def __init__(self):
        super().__init__()
        self.__initialise_registries()

    def __initialise_registries(
            self) \
            -> None:
        self.set_registry(
            name=RegistryNamesEnums.RAW_DATA,
            registry=RawDataRegistries(
                owning_universe=self))
        self.set_registry(
            name=RegistryNamesEnums.TRANSACTIONS,
            registry=TransactionRegistries(
                owning_universe=self))

Enums for registry names, register names, and configuration keys live in {pipeline_name}/objects/enums/.

---

Standard Test Pattern

Stage Unit Test

Build a Universe with test data in registers, run the stage orchestrator, assert on Universe state afterwards. No mocking of other stages.

def test_1c_collect_populates_raw_data_register():
    universe = \
        AccountsPipelineUniverses()
    # inject test config
    universe.set_configuration(
        name=ConfigKeysEnums.INPUT_PATH,
        value='tests/fixtures/sample.csv')

    orchestrate_accounts_pipeline_1c_collect(
        universe=universe)

    raw_register = \
        universe \
        .get_registry(RegistryNamesEnums.RAW_DATA) \
        .get_register(RegisterNamesEnums.CSV_ROWS)
    assert raw_register.has_content()
    assert len(raw_register.get_content()) > 0

E2E Test

Calls the full application entry point. Setup/teardown fixtures manage test data:

# tests/universal/e2e/test_{domain_name}_pipeline.py

def test_{domain_name}_b_clearer_pipeline_b_application(
        e2e_test_setup,
        e2e_test_teardown):
    run_{domain_name}_b_clearer_pipeline_b_application()
    # assert expected output files or database records exist
    assert True

---

Verification Checklist

Beyond ruff, mypy, pytest, verify after implementation:

• [ ] Each stage B-unit file is in {pipeline_name}/objects/b_units/{pipeline_name}_{stage_name}/
• [ ] Each stage orchestrator is in orchestrators/stages/
• [ ] @run_and_log_function() applied to pipeline runners and stage orchestrators
• [ ] create_and_run_b_unit() used to instantiate and run all B-units (not direct instantiation)
• [ ] Universe created in stage orchestrator (or pipeline orchestrator) — not inside individual B-units
• [ ] Interop service imports only in 1c_collect and 5r_reuse B-units
• [ ] BIE factories only in bie/ folder (if BIE domain used)
• [ ] No cross-stage imports (stage A B-units do not import stage B B-units)
• [ ] Configuration values read from Universe, not hardcoded in B-units
• [ ] Each stage has at least one independent unit test
• [ ] E2E test runs the full application entry point successfully