Autocompletion of ioc and version (#60)

* More general ioc versions function

* Add autocomplete skeleton

* More skeleton code

* Implement autocomplete

* Linting

* Ioc autocomplete cached (#2)

* Attempt caching for autocomplete

* Create available IOC & version graph

* Add catch for version autocomplete

* Remove caching from ioc instances call

* Handle --repo option

* Implement unique cache per domain repo

* Ignore files present in iocs/

* Add workaround for path autocompletion

* Add kubectl autocompletions

* Change ec logs autocomplete to running iocs

* Add some typing + error handling for autocomplete

* Ignore typer deprecated warning

* Make mypy happier

* Linting

* Make autocomplete keyerror behavior consistant

* Autocomplete with error rather than files

* fix instances test

* Fix f-string for failed test

* Include directory context manager addition

---------

Co-authored-by: Giles Knap <[email protected]>

pyproject.toml

@@ -66,7 +66,11 @@ addopts = """
     --cov=epics_containers_cli --cov-report term --cov-report xml:cov.xml
     """
 # https://iscinumpy.gitlab.io/post/bound-version-constraints/#watch-for-warnings
-filterwarnings = ["error", "ignore::pytest_cov.plugin.CovDisabledWarning"]
+filterwarnings = [
+    "error",
+    "ignore::pytest_cov.plugin.CovDisabledWarning",
+    "ignore:'autocompletion' is renamed to 'shell_complete'. The old name is deprecated and will be removed in Click 8.1. See the docs about 'Parameter' for information about new behavior.:DeprecationWarning:typer",
+    ]
 # Doctest python code in docs, python code in src docstrings, test functions in tests
 testpaths = "docs src tests"
 

src/epics_containers_cli/git.py

@@ -5,7 +5,7 @@
 import os
 import re
 from pathlib import Path
-from typing import Optional, Tuple
+from typing import Dict, Optional, Tuple
 
 import typer
 
@@ -110,38 +110,47 @@ def repo2registry(repo_name: str) -> str:
     return registry
 
 
-def versions(beamline_repo: str, ioc_name: str, folder: Path):
+def create_ioc_graph(beamline_repo: str, folder: Path) -> Dict:
     """
-    determine the versions of an IOC instance by discovering the tags in the
-    beamline repo at which changes to the instance were made since the last
-    tag
+    return a dictionary of the available IOCs (by discovering the children
+    to the iocs/ folder in the beamline repo) as well as a list of the corresponing
+    available versions for each IOC (by discovering the tags in the beamline repo at
+    which changes to the instance were made since the last tag) and the respective
+    list of available versions
     """
-    check_beamline_repo(beamline_repo)
-    typer.echo(f"Available instance versions for {ioc_name}:")
+    ioc_graph = {}
 
+    check_beamline_repo(beamline_repo)
     run_command(f"git clone {beamline_repo} {folder}", interactive=False)
-
-    ioc_name = Path(ioc_name).name
+    path_list = os.listdir(os.path.join(folder, "iocs"))
+    ioc_list = [
+        path for path in path_list if os.path.isdir(os.path.join(folder, "iocs", path))
+    ]
 
     with chdir(folder):  # From python 3.11 can use contextlib.chdir(folder)
-        result = str(run_command("git tag", interactive=False))
-        log.debug(f"checking these tags for changes in the instance: {result}")
-
-        count = 0
-        tags = result.split("\n")
-        for tag in tags:
-            if tag == "":
-                continue
-            cmd = f"git diff --name-only {tag} {tag}^"
-            result = str(run_command(cmd, interactive=False))
-
-            if ioc_name in result:
-                typer.echo(f"  {tag}")
-                count += 1
-
-        if count == 0:
-            # also look to see if the first tag was when the instance was created
-            cmd = f"git diff --name-only {tags[0]} $(git hash-object -t tree /dev/null)"
-            result = str(run_command(cmd, interactive=False))
-            if ioc_name in result:
-                typer.echo(f"  {tags[0]}")
+        for ioc_name in ioc_list:
+            ioc_name = Path(ioc_name).name
+            result = str(run_command("git tag", interactive=False))
+            log.debug(f"checking these tags for changes in the instance: {result}")
+
+            version_list = []
+            tags = result.split("\n")
+
+            for tag in tags:
+                if tag == "":
+                    continue
+                cmd = f"git diff --name-only {tag} {tag}^"
+                result = str(run_command(cmd, interactive=False))
+                if ioc_name in result:
+                    version_list.append(tag)
+
+            if not version_list:
+                # also look to see if the first tag was when the instance was created
+                cmd = f"git diff --name-only {tags[0]} $(git hash-object -t tree /dev/null)"
+                result = str(run_command(cmd, interactive=False))
+                if ioc_name in result:
+                    version_list.append(tags[0])
+
+            ioc_graph[ioc_name] = version_list
+
+    return ioc_graph

src/epics_containers_cli/globals.py

@@ -45,6 +45,12 @@ def __str__(self):
 IOC_NAME = "test-ioc"
 # namespace name for deploying IOC instances into the local podman/docker
 LOCAL_NAMESPACE = "local"
+# location for caching
+CACHE_ROOT = os.path.expanduser("~/.cache/ec-cli/")
+# available ioc cache
+IOC_CACHE = "ioc_cache.json"
+# cache expiry time in seconds
+CACHE_EXPIRY = 15
 
 # these should be set to 0 or 1 in the environment - blank is treated as false
 # Enable debug logging in all ec commands

src/epics_containers_cli/ioc/ioc_autocomplete.py

@@ -0,0 +1,140 @@
+import json
+import os
+import time
+import urllib
+from pathlib import Path
+from tempfile import mkdtemp
+from typing import List
+
+import typer
+
+from epics_containers_cli.git import create_ioc_graph
+from epics_containers_cli.globals import (
+    CACHE_EXPIRY,
+    CACHE_ROOT,
+    IOC_CACHE,
+    LOCAL_NAMESPACE,
+)
+from epics_containers_cli.ioc.k8s_commands import check_namespace
+from epics_containers_cli.logging import log
+from epics_containers_cli.shell import run_command
+
+
+def url_encode(in_string: str) -> str:
+    return urllib.parse.quote(in_string, safe="")
+
+
+def cache_dict(cache_folder: str, cached_file: str, data_struc: dict) -> None:
+    cache_dir = os.path.join(CACHE_ROOT, cache_folder)
+    if not os.path.exists(cache_dir):
+        os.makedirs(cache_dir)
+
+    cache_path = os.path.join(cache_dir, cached_file)
+    with open(cache_path, "w") as f:
+        f.write(json.dumps(data_struc, indent=4))
+
+
+def read_cached_dict(cache_folder: str, cached_file: str) -> dict:
+    cache_path = os.path.join(CACHE_ROOT, cache_folder, cached_file)
+    read_dict = {}
+
+    # Check cache if available
+    if os.path.exists(cache_path):
+        # Read from cache if not stale
+        if (time.time() - os.path.getmtime(cache_path)) < CACHE_EXPIRY:
+            with open(cache_path) as f:
+                read_dict = json.load(f)
+
+    return read_dict
+
+
+def fetch_ioc_graph(beamline_repo: str) -> dict:
+    ioc_graph = read_cached_dict(url_encode(beamline_repo), IOC_CACHE)
+    if not ioc_graph:
+        ioc_graph = create_ioc_graph(beamline_repo, Path(mkdtemp()))
+        cache_dict(url_encode(beamline_repo), IOC_CACHE, ioc_graph)
+
+    return ioc_graph
+
+
+def avail_IOCs(ctx: typer.Context) -> List[str]:
+    params = ctx.parent.parent.params  # type: ignore
+    beamline_repo = params["repo"] or os.environ.get("EC_DOMAIN_REPO", "")
+
+    # This block prevents getting a stack trace during autocompletion
+    try:
+        ioc_graph = fetch_ioc_graph(beamline_repo)
+        return list(ioc_graph.keys())
+    except typer.Exit:
+        return [" "]
+    except Exception:
+        log.error("Error")
+        return [" "]
+
+
+def avail_versions(ctx: typer.Context) -> List[str]:
+    params = ctx.parent.parent.params  # type: ignore
+    beamline_repo = params["repo"] or os.environ.get("EC_DOMAIN_REPO", "")
+    ioc_name = ctx.params["ioc_name"]
+
+    # This block prevents getting a stack trace during autocompletion
+    try:
+        ioc_graph = fetch_ioc_graph(beamline_repo)
+        ioc_versions = ioc_graph[ioc_name]
+        return ioc_versions
+    except KeyError:
+        log.error("IOC not found")
+        return [" "]
+    except typer.Exit:
+        return [" "]
+    except Exception:
+        log.error("Error")
+        return [" "]
+
+
+def force_plain_completion() -> List[str]:
+    return []
+
+
+def running_iocs(ctx: typer.Context) -> List[str]:
+    params = ctx.parent.parent.params  # type: ignore
+    namespace = params["namespace"] or os.environ.get("EC_K8S_NAMESPACE", "")
+
+    # This block prevents getting a stack trace during autocompletion
+    try:
+        if namespace == LOCAL_NAMESPACE:
+            # Not yet implemented
+            return []
+        else:
+            check_namespace(namespace)
+            columns = "-o custom-columns=IOC_NAME:metadata.labels.app"
+            command = f"kubectl -n {namespace} get pod -l is_ioc==True {columns}"
+            ioc_list = str(run_command(command, interactive=False)).split()[1:]
+            return ioc_list
+    except typer.Exit:
+        return [" "]
+    except Exception:
+        log.error("Error")
+        return [" "]
+
+
+def all_iocs(ctx: typer.Context) -> List[str]:
+    params = ctx.parent.parent.params  # type: ignore
+    namespace = params["namespace"] or os.environ.get("EC_K8S_NAMESPACE", "")
+
+    # This block prevents getting a stack trace during autocompletion
+    try:
+        if namespace == LOCAL_NAMESPACE:
+            # Not yet implemented
+            return []
+        else:
+            check_namespace(namespace)
+            columns = "-o custom-columns=DEPLOYMENT:metadata.labels.app"
+            command = f"kubectl -n {namespace} get deploy -l is_ioc==True {columns}"
+            ioc_list = str(run_command(command, interactive=False)).split()[1:]
+            return ioc_list
+    except typer.Exit:
+        return [" "]
+    except Exception:
+        log.error("Error")
+        return [" "]