🔖 0.12.1 (#183)

* Add utils.is_loading_pipeline() to check if pipeline is loading

* 🔖 0.12.1

README.md

@@ -64,12 +64,12 @@ if __name__ == "__main__":
 06-09 23:15:29 I core                  _  ____/__/ /  _  ____/_  /___  _  /|  /
 06-09 23:15:29 I core                  /_/     /___/  /_/     /_____/  /_/ |_/
 06-09 23:15:29 I core
-06-09 23:15:29 I core                              version: 0.11.0
+06-09 23:15:29 I core                              version: 0.12.1
 06-09 23:15:29 I core
 06-09 23:15:29 I core    ╔═══════════════════════════════════════════════════╗
 06-09 23:15:29 I core    ║                            MYPIPELINE                            ║
 06-09 23:15:29 I core    ╚═══════════════════════════════════════════════════╝
-06-09 23:15:29 I core    plugins         : verbose v0.7.0
+06-09 23:15:29 I core    plugins         : verbose v0.9.0
 06-09 23:15:29 I core    # procs         : 2
 06-09 23:15:29 I core    profile         : default
 06-09 23:15:29 I core    outdir          : /home/pwwang/github/pipen/MyPipeline-output

docs/CHANGELOG.md

@@ -1,5 +1,9 @@
 # Change Log
 
+## 0.12.1
+
+- Add utils.is_loading_pipeline() to check if pipeline is loading
+
 ## 0.12.0
 
 - ✨ Add utils.load_pipeline() to load pipeline

pipen/utils.py

@@ -56,6 +56,8 @@
     from .proc import Proc
     from .procgroup import ProcGroup
 
+LOADING_ARGV0 = "@pipen"
+
 
 class RichHandler(_RichHandler):
     """Subclass of rich.logging.RichHandler, showing log levels as a single
@@ -689,7 +691,7 @@ async def load_pipeline(
         )
 
     old_argv = sys.argv
-    sys.argv = ["@pipen"] + list(cli_args)
+    sys.argv = [LOADING_ARGV0] + list(cli_args)
     try:
         # Initialize the pipeline so that the arguments definied by
         # other plugins (i.e. pipen-args) to take in place.
@@ -703,3 +705,13 @@ async def load_pipeline(
         sys.argv = old_argv
 
     return pipeline
+
+
+def is_loading_pipeline() -> bool:
+    """Check if we are loading the pipeline
+
+    Returns:
+        True if we are loading the pipeline (argv[0] == "@pipen"),
+        otherwise False
+    """
+    return sys.argv[0] == LOADING_ARGV0

pipen/version.py

@@ -1,3 +1,3 @@
 """Provide version of pipen"""
 
-__version__ = "0.12.0"
+__version__ = "0.12.1"

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.masonry.api"
 
 [tool.poetry]
 name = "pipen"
-version = "0.12.0"
+version = "0.12.1"
 description = "A pipeline framework for python"
 authors = [ "pwwang <[email protected]>",]
 license = "MIT"

tests/helpers.py

@@ -7,6 +7,7 @@
 
 import pytest
 from pipen import Proc, Pipen, plugin
+from pipen.utils import is_loading_pipeline
 
 
 class SimpleProc(Proc):
@@ -163,6 +164,8 @@ class DirOutputProc(Proc):
 class SimplePlugin:
     @plugin.impl
     async def on_init(pipen):
+        if getattr(pipen.__class__, "loading", False):
+            assert is_loading_pipeline()
         print("SimplePlugin")
 
 
@@ -199,6 +202,13 @@ def pipen_with_plugin(tmp_path):
     return pipen_simple
 
 
+class PipenIsLoading(Pipen):
+    name = "PipenIsLoading"
+    loading = True
+    plugins = [SimplePlugin()]
+    starts = SimpleProc
+
+
 @pytest.fixture
 def infile(tmp_path):
     out = tmp_path / "infile"

tests/test_utils.py

@@ -170,12 +170,7 @@ class P1(Proc):
     assert pipeline.name == "PG"
 
     # Pipen
-    obj = _get_obj_from_spec(f"{HERE}/helpers.py:pipen")
-    proc = _get_obj_from_spec(f"{HERE}/helpers.py:SimpleProc")
-    # Use the original function instead of fixture
-    p = obj.__wrapped__(tmp_path).__class__
-    p.starts = [proc]
-    pipeline = await load_pipeline(p, name="LoadedPipeline")
-    assert pipeline.name == "LoadedPipeline"
+    pipeline = await load_pipeline(f"{HERE}/helpers.py:PipenIsLoading")
+    assert pipeline.name == "PipenIsLoading"
     assert pipeline.starts[0].name == "SimpleProc"
     assert len(pipeline.procs) == 1