Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: pull out meta properties of CQM protocols without importing file #140

Merged
merged 1 commit into from
Oct 18, 2024
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
75 changes: 52 additions & 23 deletions canvas_cli/apps/plugin/plugin.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,5 @@
import importlib
import ast
import json
import os
import sys
import tarfile
import tempfile
from pathlib import Path
Expand Down Expand Up @@ -72,28 +70,62 @@ def _get_name_from_metadata(host: str, token: str, package: Path) -> Optional[st
return metadata.get("name")


def _get_meta_class(text: str, classname: str) -> ast.ClassDef | None:
tree = ast.parse(text)
for node in ast.walk(tree):
if isinstance(node, ast.ClassDef) and node.name == classname:
class_def = node
for b in class_def.body:
if isinstance(b, ast.ClassDef) and b.name == "Meta":
return b
return None


def _get_meta_properties(protocol_path: Path, classname: str) -> dict[str, str]:
meta: dict[str, str] = {}

if not protocol_path.exists():
return meta

meta_class = _get_meta_class(protocol_path.read_text(), classname)
if not meta_class:
return meta

for meta_b in meta_class.body:
if not isinstance(meta_b, ast.Assign):
continue
target_id = next((t.id for t in meta_b.targets if isinstance(t, ast.Name)), None)
if not target_id:
continue
if isinstance(meta_b.value, ast.Constant):
value = meta_b.value.value
elif isinstance(meta_b.value, ast.List):
value = [e.value for e in meta_b.value.elts]
else:
value = None
meta[target_id] = value

return meta


def _get_protocols_with_new_cqm_properties(
protocol_classes: List[dict[str, Any]]
protocol_classes: List[dict[str, Any]], plugin: Path
) -> List[dict[str, Any]] | None:
"""Extract the meta properties of any ClinicalQualityMeasure Protocols included in the plugin if they have changed."""

def get_new_meta_properties(protocol_class: dict[str, str]) -> dict[str, str]:
mod, classname = protocol_class["class"].split(":")
module = importlib.import_module(mod)
_class = getattr(module, classname)
if not hasattr(_class, "_meta") or _class._meta() == protocol_class.get("meta"):
return {}
return {"meta": _class._meta()}

has_updates = False
new_protocols = []
protocol_props = []
for p in protocol_classes:
m = get_new_meta_properties(p)
new_protocols.append(p | m)
if m:
mod, classname = p["class"].split(":")
mod = mod.replace(f"{plugin.name}.", "").replace(".", "/") + ".py"
p_path: Path = plugin / mod
meta = _get_meta_properties(p_path, classname)
if meta and meta != p.get("meta"):
has_updates = True
protocol_props.append(p | {"meta": meta})
else:
protocol_props.append(p)

return new_protocols if has_updates else None
return protocol_props if has_updates else None


def get_base_plugin_template_path() -> Path:
Expand Down Expand Up @@ -331,11 +363,8 @@ def validate_manifest(

try:
manifest_json = json.loads(manifest.read_text())

sys.path.append(str(plugin_name.parent.absolute()))
if new_protocols := _get_protocols_with_new_cqm_properties(
manifest_json.get("components", {}).get("protocols", [])
):
protocols = manifest_json.get("components", {}).get("protocols", [])
if new_protocols := _get_protocols_with_new_cqm_properties(protocols, plugin_name):
print(
f"Updating the CANVAS_MANIFEST.json file for {plugin_name} with CQM meta properties"
)
Expand Down
Loading