CI: Check of async code is up-to-date

Synopsis:

  # Check if generated code is up-to-date.
  python script/generate_async.py check

  # Generate code.
  python script/generate_async.py format

docs/development.md

@@ -2,36 +2,36 @@
 
 ## Sandbox
 In order to create a development sandbox, you may want to follow this list of
-commands. When you see the software tests succeed, you should be ready to start
-hacking.
-
+commands.
 ```shell
 git clone https://github.com/panodata/grafana-client
 cd grafana-client
 python3 -m venv .venv
 source .venv/bin/activate
 pip install --editable=.[test,develop]
+```
 
+## Software Tests
+When you see the software tests succeed, you should be ready to start
+hacking.
+```shell
 # Run all tests.
 poe test
 
 # Run specific tests.
 python -m unittest -k preference -vvv
 ```
 
-### Formatting
-
-Before creating a PR, you can run `poe format`, in order to resolve code style issues.
-
-### Async code
-
-If you update any piece of code in `grafana_client/elements/*`, please run:
-
-```
-python script/generate_async.py
+## Code Formatting
+Before submitting a PR, please format the code, in order to invoke the async
+translation program and to resolve code style issues.
+```shell
+poe format
 ```
 
-Do not edit files in `grafana_client/elements/_async/*` manually.
+The async translation program populates the `grafana_client/elements/_async`
+folder automatically. Please do not edit files there manually.
+
 
 ## Run Grafana
 ```

pyproject.toml

@@ -40,10 +40,12 @@ extend-exclude = [
 
 [tool.poe.tasks]
 format = [
+  {cmd="script/generate_async.py format"},
   {cmd="black ."},
   {cmd="isort ."},
 ]
 lint = [
+  {cmd="script/generate_async.py check"},
   {cmd="ruff ."},
   {cmd="black --check ."},
   {cmd="isort --check ."},

script/generate_async.py

@@ -1,25 +1,71 @@
+#!/usr/bin/env python
 """
-Script to automatically generate grafana asynchronous code from the
+Program to automatically generate asynchronous code from the
 synchronous counterpart.
-What does this program do:
-- For all module in grafana_client/elements/*.py excepted base and __init__
-Inject async/await keyword in all available methods / client interactions.
-- Then detect no longer needed module for removal.
-- Finally generate the _async top level code based on the elements/__init__.py one.
+
+Synopsis:
+
+  # Check if generated code is up-to-date.
+  python script/generate_async.py check
+
+  # Generate code.
+  python script/generate_async.py format
+
+What does this program does:
+- For each module in `grafana_client/elements/*.py`, except `base` and `__init__`:
+  Inject async/await keyword in all available methods / client interactions.
+- Detect modules no longer in use, and remove them.
+- Generate the async top level code based on `elements/__init__.py`.
 """
 
 import os
 import re
+import shutil
 import subprocess
+import sys
+import tempfile
 from glob import glob
+from pathlib import Path
 
 BASE_PATH = ".." if os.path.exists("../grafana_client") else "."
 
-if __name__ == "__main__":
+SOURCE = f"{BASE_PATH}/grafana_client/elements"
+TARGET = f"{BASE_PATH}/grafana_client/elements/_async"
+
+
+def msg(text: str):
+    status = ""
+    if text.lower().startswith("info"):
+        status = "\033[92;1m"
+    elif text.lower().startswith("error"):
+        status = "\033[91;1m"
+    clear = "\033[0m"
+    return f"{status}{text}{clear}"
+
+
+def run(action: str):
+
+    run_check = action == "check" or False
+    run_format = action == "format" or False
+    assert run_check or run_format, "Wrong or missing action, use either `check` or `format`."
+
     module_processed = []
     module_generated = []
 
-    for module_path in glob(f"{BASE_PATH}/grafana_client/elements/*.py"):
+    source = SOURCE
+    target = TARGET
+
+    tmpdir = None
+    if run_check:
+        # Check needs to run in a local subdirectory, so the same
+        # code style settings apply like for the main code base.
+        tmp_parent = Path().cwd() / "tmp"
+        tmpdir = Path(tempfile.mkdtemp(prefix=f"{tmp_parent}/foo"))
+        target = str(tmpdir)
+
+    Path(target).mkdir(exist_ok=True)
+
+    for module_path in glob(f"{source}/*.py"):
         if "__init__.py" in module_path or "base.py" in module_path:
             continue
 
@@ -39,33 +85,31 @@
         module_dump = module_dump.replace("= self.", "= await self.")
 
         module_processed.append(module_path)
-        target_path = module_path.replace("elements/", "elements/_async/")
+        target_path = module_path.replace(source, target)
         module_generated.append(target_path)
 
         print(f"Writing to {target_path}...")
 
-        with open(module_path.replace("elements/", "elements/_async/"), "w") as fp:
+        with open(module_path.replace(source, target), "w") as fp:
             fp.write(module_dump)
 
     relevant_modules = [os.path.basename(_) for _ in module_processed]
     existing_modules = [
-        os.path.basename(_)
-        for _ in glob(f"{BASE_PATH}/grafana_client/elements/_async/*.py")
-        if "base.py" not in _ and "__init__.py" not in _
+        os.path.basename(_) for _ in glob(f"{target}/*.py") if "base.py" not in _ and "__init__.py" not in _
     ]
 
     remove_module_count = 0
 
     for existing_module in existing_modules:
         if existing_module not in relevant_modules:
             print(f"Removing module {existing_module}...")
-            os.remove(f"{BASE_PATH}/grafana_client/elements/_async/{existing_module}")
+            os.remove(f"{target}/{existing_module}")
             remove_module_count += 1
 
     if not remove_module_count:
         print("No modules to remove.. pursuing..")
 
-    with open(f"{BASE_PATH}/grafana_client/elements/__init__.py", "r") as fp:
+    with open(f"{source}/__init__.py", "r") as fp:
         top_level_content = fp.read()
 
     print("Updating _async top level import content")
@@ -99,7 +143,29 @@
 
         top_level_content_patch.append(line)
 
-    with open(f"{BASE_PATH}/grafana_client/elements/_async/__init__.py", "w") as fp:
+    with open(f"{target}/__init__.py", "w") as fp:
         fp.write("\n".join(top_level_content_patch) + "\n")
 
-    subprocess.call(["poe", "format"])
+    subprocess.call(["black", target])
+    subprocess.call(["isort", target])
+
+    exitcode = 0
+
+    if run_check:
+        command = f"diff -u {TARGET} {target}"
+        exitcode = os.system(command)
+        if exitcode == 0:
+            print(msg("INFO: Async code up-to-date. Excellent."))
+        else:
+            print(msg("ERROR: Async code not up-to-date. Please run `poe format`."))
+            exitcode = 2
+
+    if tmpdir:
+        shutil.rmtree(tmpdir)
+
+    sys.exit(exitcode)
+
+
+if __name__ == "__main__":
+    subcommand = sys.argv[1:] and sys.argv[1] or None
+    run(subcommand)

tmp/.gitignore

@@ -0,0 +1,3 @@
+*
+!.gitignore
+!.gitkeep