Generate requirements public api

src/ewoks/__init__.py

@@ -3,6 +3,7 @@
 
 from .bindings import convert_graph  # noqa: F401
 from .bindings import execute_graph  # noqa: F401
+from .bindings import generate_requirements  # noqa: F401
 from .bindings import load_graph  # noqa: F401
 from .bindings import save_graph  # noqa: F401
 from .bindings import show_graph  # noqa: F401

src/ewoks/_requirements/__init__.py

@@ -56,12 +56,28 @@ def add_requirements(
     package manager that generated them. `engine` is the name and version of the
     distribution of the engine that saves the workflow.
     """
+    requirements, name = generate_requirements(
+        manager_name=manager_name, manager_command=manager_command, engine=engine
+    )
+    graph.graph.graph["requirements"] = requirements
+    return name
+
+
+def generate_requirements(
+    manager_name: Optional[str] = None,
+    manager_command: Tuple[str, ...] = tuple(),
+    engine: Optional[Tuple[str, str]] = None,
+) -> Tuple[dict, str]:
+    """Requirements of the current python environment, as stored in the
+    `requirements` graph attribute, and the name of the package manager that
+    generated them. `engine` is the name and version of the distribution of the
+    engine that saves the workflow.
+    """
     manager = get_manager(manager_name=manager_name, manager_command=manager_command)
     requirements = manager.gather_requirements()
     if requirements.ewoks and engine:
         requirements.ewoks.engine = EngineInfo(name=engine[0], version=engine[1])
-    graph.graph.graph["requirements"] = requirements.model_dump()
-    return manager.NAME
+    return requirements.model_dump(), manager.NAME
 
 
 def get_requirements(graph: TaskGraph) -> BaseRequirements:

src/ewoks/bindings.py

@@ -217,18 +217,13 @@ def convert_graph(
     graph = load_graph(source, inputs=inputs, **load_options)
 
     if save_requirements:
-        if not package_manager_command:
-            package_manager_command = tuple()
-        elif isinstance(package_manager_command, str):
-            package_manager_command = _split_command(package_manager_command)
-
         engine, _ = _engines.get_serialization_engine(
             destination, representation=save_options.get("representation")
         )
         manager_name = _requirements.add_requirements(
             graph,
             manager_name=package_manager_name,
-            manager_command=package_manager_command,
+            manager_command=_manager_command(package_manager_command),
             engine=_engines.get_engine_distribution(engine),
         )
         logger.info("Requirements generated with the %r package manager", manager_name)
@@ -242,6 +237,34 @@ def convert_graph(
     return save_graph(graph, destination, **save_options)
 
 
+def generate_requirements(
+    package_manager_name: Optional[str] = None,
+    package_manager_command: Union[None, str, Tuple[str, ...]] = None,
+    engine: Optional[Tuple[str, str]] = None,
+) -> dict:
+    """Requirements of the current python environment, as `convert_graph` saves
+    them in the `requirements` graph attribute. `engine` is the name and version of
+    the distribution of the engine that saves the workflow.
+    """
+    requirements, manager_name = _requirements.generate_requirements(
+        manager_name=package_manager_name,
+        manager_command=_manager_command(package_manager_command),
+        engine=engine,
+    )
+    logger.info("Requirements generated with the %r package manager", manager_name)
+    return requirements
+
+
+def _manager_command(
+    package_manager_command: Union[None, str, Tuple[str, ...]],
+) -> Tuple[str, ...]:
+    if not package_manager_command:
+        return tuple()
+    if isinstance(package_manager_command, str):
+        return _split_command(package_manager_command)
+    return tuple(package_manager_command)
+
+
 def show_graph(
     source,
     inputs: Optional[List[dict]] = None,

src/ewoks/tests/test_generate_requirements.py

@@ -0,0 +1,18 @@
+from ewoks import convert_graph
+from ewoks import generate_requirements
+
+
+def test_generate_requirements():
+    requirements = generate_requirements(engine=("ewoksorange", "1.2.3"))
+    assert requirements["ewoks"]["engine"] == {
+        "name": "ewoksorange",
+        "version": "1.2.3",
+    }
+    assert requirements["distributions"]
+
+
+def test_generate_requirements_like_convert():
+    graph = {"graph": {"id": "test"}, "nodes": [], "links": []}
+    converted = convert_graph(graph, None, save_options={"representation": "json_dict"})
+    requirements = generate_requirements()
+    assert requirements.keys() == converted["graph"]["requirements"].keys()