Support initial context

bc2/core/pipeline.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Any
 
 from pydantic import BaseModel
 
@@ -33,19 +34,26 @@ def validate(self, runtime_config: RuntimeConfig):
         """Validate the pipeline configuration."""
         self.pipeline.validate(runtime_config)
 
-    def run(self, runtime_config: RuntimeConfig | None = None) -> Context:
+    def run(
+        self,
+        runtime_config: RuntimeConfig | None = None,
+        context: Context | dict[str, Any] | None = None,
+    ) -> Context:
         """Run the pipeline.
 
         Args:
             runtime_config: The runtime configuration.
             Most values in the runtime config are dependent on the pipeline.
             The `debug` flag is interpretted globally.
+            context: Initial context object to use for the pipeline run.
 
         Returns:
             The context object created during the pipeline run.
+            Note if an initial context is provided, it will be copied but not modified;
+            a new reference is returned at the end of the run.
         """
         runtime_config = runtime_config or {}
-        ctx = Context()
+        ctx = Context(context or {})
         ctx.debug = runtime_config.get("debug", False)
         ctx.errors = list[Exception]()
         if ctx.debug:

bc2/core/test_pipeline.py

@@ -224,3 +224,34 @@ def test_pipeline_invalid_optional_step():
         "<class 'bc2.core.common.file.MemoryFile'> is not compatible with "
         "the output type <class 'bc2.core.common.text.Text'>."
     )
+
+
+def test_pipeline_initial_context():
+    cfg = PipelineConfig.model_validate(
+        {
+            "pipe": [
+                {"engine": "in:memory"},
+                {"engine": "extract:raw"},
+                {"engine": "redact:noop", "delimiters": ["[", "]"]},
+                {"engine": "render:text"},
+                {"engine": "out:memory"},
+            ],
+        }
+    )
+    # Run the pipeline.
+    pipe = Pipeline(cfg)
+
+    in_buf = io.BytesIO(b"Hello, world!")
+    out_buf = io.BytesIO()
+    opts = {
+        "debug": True,
+        "in": {
+            "buffer": in_buf,
+        },
+        "out": {
+            "buffer": out_buf,
+        },
+    }
+    ctx = pipe.run(runtime_config=opts, context={"foo": "bar"})
+    assert ctx.foo == "bar"
+    assert ctx.debug is True