Skip to content

pipeline

Classes

PipelineRenderer

Bases: BaseJinjaRenderer[Pipeline, RenderInputsSchema]

Renders a basic Jupyter notebook from a pipeline.

Source code in kiara_plugin/jupyter/renderers/pipeline.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
class PipelineRenderer(BaseJinjaRenderer[Pipeline, RenderInputsSchema]):
    """Renders a basic Jupyter notebook from a pipeline."""

    _renderer_name = "pipeline_notebook"

    def retrieve_supported_render_sources(self) -> str:
        return "pipeline"

    def retrieve_supported_render_targets(cls) -> Union[Iterable[str], str]:
        return "jupyter_notebook"

    def retrieve_source_transformers(self) -> Iterable[SourceTransformer]:
        return [PipelineTransformer(kiara=self._kiara)]

    def retrieve_jinja_env(self) -> JinjaEnv:

        jinja_env = JinjaEnv(template_base="kiara_plugin.jupyter")
        return jinja_env

    def get_template(self, render_config: RenderInputsSchema) -> Template:

        return self.get_jinja_env().get_template(
            "pipeline/workflow_tutorial/jupyter_notebook.ipynb.j2"
        )

    def assemble_render_inputs(
        self, instance: Any, render_config: RenderInputsSchema
    ) -> Mapping[str, Any]:

        inputs = render_config.dict()
        inputs["pipeline"] = instance
        return inputs

    def _post_process(self, rendered: str) -> str:

        is_notebook = True
        if is_notebook:
            import jupytext

            notebook = jupytext.reads(rendered, fmt="py:percent")
            converted = jupytext.writes(notebook, fmt="notebook")
            return converted
        else:
            try:
                import black
                from black import Mode  # type: ignore

                cleaned = black.format_str(rendered, mode=Mode())
                return cleaned

            except Exception as e:
                log_message(
                    f"Could not format python code, 'black' not in virtual environment: {e}."
                )
                return rendered

Functions

retrieve_supported_render_sources() -> str
Source code in kiara_plugin/jupyter/renderers/pipeline.py 
17
18
def retrieve_supported_render_sources(self) -> str:
    return "pipeline"
retrieve_supported_render_targets() -> Union[Iterable[str], str]
Source code in kiara_plugin/jupyter/renderers/pipeline.py 
20
21
def retrieve_supported_render_targets(cls) -> Union[Iterable[str], str]:
    return "jupyter_notebook"
retrieve_source_transformers() -> Iterable[SourceTransformer]
Source code in kiara_plugin/jupyter/renderers/pipeline.py 
23
24
def retrieve_source_transformers(self) -> Iterable[SourceTransformer]:
    return [PipelineTransformer(kiara=self._kiara)]
retrieve_jinja_env() -> JinjaEnv
Source code in kiara_plugin/jupyter/renderers/pipeline.py 
26
27
28
29
def retrieve_jinja_env(self) -> JinjaEnv:

    jinja_env = JinjaEnv(template_base="kiara_plugin.jupyter")
    return jinja_env
get_template(render_config: RenderInputsSchema) -> Template
Source code in kiara_plugin/jupyter/renderers/pipeline.py 
31
32
33
34
35
def get_template(self, render_config: RenderInputsSchema) -> Template:

    return self.get_jinja_env().get_template(
        "pipeline/workflow_tutorial/jupyter_notebook.ipynb.j2"
    )
assemble_render_inputs(instance: Any, render_config: RenderInputsSchema) -> Mapping[str, Any]
Source code in kiara_plugin/jupyter/renderers/pipeline.py 
37
38
39
40
41
42
43
def assemble_render_inputs(
    self, instance: Any, render_config: RenderInputsSchema
) -> Mapping[str, Any]:

    inputs = render_config.dict()
    inputs["pipeline"] = instance
    return inputs