PipelineProject

PipelineProject is the project-scoped ingest surface. You get one from Pipeline.project() or Pipeline.project_from(). Its single public method, ingest(), has two distinct call shapes.

from optixlog.pipeline import PipelineProject

PipelineProject.id

The project ID string this instance is bound to.

id: str

The IngestibleNode protocol

Before documenting ingest(), it helps to understand what the data argument must be.

Any object that implements to_payload() satisfies the IngestibleNode protocol:

class IngestibleNode(Protocol):
    def to_payload(self) -> dict[str, JSONValue]: ...

You will normally use generated node classes (e.g. SimulationNode, MeasurementNode) produced by optixlog generate. Each generated class is a @dataclass(frozen=True, kw_only=True) with a to_payload() method. You can also write your own class that implements to_payload().

The node's class name (e.g. "SimulationNode") is sent to the server as the node_type field, which the server resolves against the project's workflow schema.

PipelineProject.ingest(...) — overload 1: prompt

Ingest a node and let the server infer graph edges from a natural-language prompt.

Signature

@overload
def ingest(
    self,
    *,
    data: IngestibleNode,
    prompt: str,
) -> IngestResult: ...

Parameters

Returns — IngestResult: contains project_id, node_id, created flag, and the resolved origin_nodes/forward_nodes edge lists.

Side effects — Makes a POST network request (v0.ingest) that creates a new workflow node or updates an existing one in the project graph, and wires its edges as inferred from prompt.

Raises

• AuthenticationError — the API key was missing, malformed, or rejected (HTTP 401/403).
• ValidationError — the server rejected the payload as invalid (HTTP 400/422), e.g. an unrecognised node_type for this project.
• OptixLogError — any other HTTP error or transport failure.

Example

from optixlog import SDKClient
from optixlog.pipeline import Pipeline
from optixlog_gen import SimulationNode

client = SDKClient(base_url="https://optixlog.leidos.com", api_key="sk-opt-...")
project = Pipeline(client).project("proj_grating_7f3a")

result = project.ingest(
    data=SimulationNode(config={"mesh": "fine"}, solver="fdtd", wavelength_nm=1550.0),
    prompt="FDTD simulation of grating coupler at 1550 nm, follows the layout step.",
)
print(f"Node {'created' if result.created else 'updated'}: {result.node_id}")

PipelineProject.ingest(...) — overload 2: explicit edges

Ingest a node with explicit predecessor and successor node IDs instead of a prompt.

Signature

@overload
def ingest(
    self,
    *,
    data: IngestibleNode,
    origin_nodes: Sequence[str],
    forward_nodes: Sequence[str],
) -> IngestResult: ...

Parameters

Returns — IngestResult: contains project_id, node_id, created flag, and the resolved origin_nodes/forward_nodes edge lists (as EntityRef objects).

Side effects — Makes a POST network request (v0.ingest) that creates a new workflow node or updates an existing one in the project graph, and wires its edges to the specified node IDs.

Raises

• AuthenticationError — the API key was missing, malformed, or rejected (HTTP 401/403).
• ValidationError — the server rejected the payload as invalid (HTTP 400/422), e.g. an unrecognised node_type or a referenced node ID that does not exist.
• OptixLogError — any other HTTP error or transport failure.

Example

from optixlog import SDKClient
from optixlog.pipeline import Pipeline
from optixlog_gen import MeasurementNode

client = SDKClient(base_url="https://optixlog.leidos.com", api_key="sk-opt-...")
project = Pipeline(client).project("proj_grating_7f3a")

# Wire this measurement node as a successor of an existing simulation node
result = project.ingest(
    data=MeasurementNode(instrument="OSA", samples=[0.92, 0.93, 0.91]),
    origin_nodes=["node_sim_abc123"],
    forward_nodes=[],
)
print(result.node_id)
for ref in result.origin_nodes:
    print(f"  origin: {ref.id}")