Skills › Scientific Agent Skills › dnanexus-integration › inspect_dxpy.py
inspect_dxpy.py
366 lines · skills/dnanexus-integration/scripts/inspect_dxpy.py at 5f2bc03
1#!/usr/bin/env python3
2"""Inspect local dxpy symbols and signatures without authentication or network."""
4from __future__ import annotations
6import argparse
7import importlib
8import inspect
9import json
10import platform
11import re
12import sys
13from importlib.metadata import PackageNotFoundError, version
14from typing import Any, Optional
17DOCUMENTED_BASELINE = "0.415.0"
19SYMBOL_GROUPS: dict[str, list[tuple[str, str]]] = {
20 "files": [
21 ("dxpy", "upload_local_file"),
22 ("dxpy", "download_dxfile"),
23 ("dxpy", "open_dxfile"),
24 ("dxpy", "upload_string"),
25 ("dxpy", "new_dxfile"),
26 ],
27 "search": [
28 ("dxpy", "find_data_objects"),
29 ("dxpy", "find_executions"),
30 ("dxpy", "find_jobs"),
31 ("dxpy", "find_analyses"),
32 ("dxpy", "find_projects"),
33 ],
34 "links_and_objects": [
35 ("dxpy", "dxlink"),
36 ("dxpy", "get_handler"),
37 ("dxpy", "describe"),
38 ("dxpy", "new_dxrecord"),
39 ("dxpy", "new_dxworkflow"),
40 ],
41 "handlers": [
42 ("dxpy", "DXFile"),
43 ("dxpy", "DXRecord"),
44 ("dxpy", "DXApplet"),
45 ("dxpy", "DXApp"),
46 ("dxpy", "DXWorkflow"),
47 ("dxpy", "DXJob"),
48 ("dxpy", "DXAnalysis"),
49 ("dxpy", "DXProject"),
50 ],
51 "exceptions": [
52 ("dxpy.exceptions", "DXError"),
53 ("dxpy.exceptions", "DXAPIError"),
54 ("dxpy.exceptions", "DXJobFailureError"),
55 ("dxpy.exceptions", "DXSearchError"),
56 ("dxpy.exceptions", "DXFileError"),
57 ("dxpy.exceptions", "DXChecksumMismatchError"),
58 ],
59}
61METHOD_GROUPS: dict[str, tuple[str, str, list[str]]] = {
62 "DXFile": (
63 "dxpy",
64 "DXFile",
65 [
66 "describe",
67 "clone",
68 "set_properties",
69 "add_tags",
70 "rename",
71 "move",
72 "close",
73 ],
74 ),
75 "DXRecord": (
76 "dxpy",
77 "DXRecord",
78 ["describe", "get_details", "set_details", "close"],
79 ),
80 "DXApplet": (
81 "dxpy",
82 "DXApplet",
83 ["describe", "run"],
84 ),
85 "DXApp": (
86 "dxpy",
87 "DXApp",
88 ["describe", "run"],
89 ),
90 "DXWorkflow": (
91 "dxpy",
92 "DXWorkflow",
93 ["describe", "run", "add_stage", "close"],
94 ),
95 "DXJob": (
96 "dxpy",
97 "DXJob",
98 ["describe", "wait_on_done", "get_output_ref", "terminate"],
99 ),
100 "DXAnalysis": (
101 "dxpy",
102 "DXAnalysis",
103 ["describe", "wait_on_done", "get_output_ref", "terminate"],
104 ),
105 "DXProject": (
106 "dxpy",
107 "DXProject",
108 [
109 "describe",
110 "list_folder",
111 "new_folder",
112 "move",
113 "remove_objects",
114 "remove_folder",
115 ],
116 ),
117}
119REQUIRED_SYMBOLS = {
120 "dxpy.upload_local_file",
121 "dxpy.download_dxfile",
122 "dxpy.open_dxfile",
123 "dxpy.find_data_objects",
124 "dxpy.find_executions",
125 "dxpy.dxlink",
126 "dxpy.get_handler",
127 "dxpy.DXFile",
128 "dxpy.DXApplet",
129 "dxpy.DXWorkflow",
130 "dxpy.DXJob",
131 "dxpy.DXAnalysis",
132 "dxpy.DXProject",
133 "dxpy.exceptions.DXAPIError",
134 "dxpy.exceptions.DXJobFailureError",
135}
138def safe_signature(obj: Any) -> Optional[str]:
139 try:
140 return str(inspect.signature(obj))
141 except (TypeError, ValueError):
142 return None
145def inspect_symbol(module_name: str, symbol_name: str) -> dict[str, Any]:
146 qualified_name = f"{module_name}.{symbol_name}"
147 try:
148 module = importlib.import_module(module_name)
149 except Exception as error:
150 return {
151 "qualified_name": qualified_name,
152 "available": False,
153 "error": f"{type(error).__name__}: {error}",
154 }
156 if not hasattr(module, symbol_name):
157 return {
158 "qualified_name": qualified_name,
159 "available": False,
160 "error": "symbol not found",
161 }
163 obj = getattr(module, symbol_name)
164 return {
165 "qualified_name": qualified_name,
166 "available": True,
167 "kind": type(obj).__name__,
168 "signature": safe_signature(obj),
169 }
172def inspect_methods(
173 module_name: str,
174 class_name: str,
175 method_names: list[str],
176) -> dict[str, Any]:
177 try:
178 cls = getattr(importlib.import_module(module_name), class_name)
179 except Exception as error:
180 return {"error": f"{type(error).__name__}: {error}"}
182 methods: dict[str, Any] = {}
183 for method_name in method_names:
184 method = getattr(cls, method_name, None)
185 methods[method_name] = {
186 "available": method is not None,
187 "signature": safe_signature(method) if method is not None else None,
188 }
189 return methods
192def numeric_version(value: str) -> tuple[int, ...]:
193 match = re.match(r"^(\d+(?:\.\d+)*)", value)
194 if match is None:
195 return ()
196 return tuple(int(part) for part in match.group(1).split("."))
199def build_report() -> dict[str, Any]:
200 try:
201 dxpy_version = version("dxpy")
202 except PackageNotFoundError:
203 return {
204 "schema_version": "1.0",
205 "python": platform.python_version(),
206 "platform": platform.platform(),
207 "documented_baseline": DOCUMENTED_BASELINE,
208 "dxpy_version": None,
209 "version_meets_baseline": False,
210 "symbols": {},
211 "methods": {},
212 "known_legacy_checks": {},
213 }
215 symbols = {
216 group: [
217 inspect_symbol(module_name, symbol_name)
218 for module_name, symbol_name in entries
219 ]
220 for group, entries in SYMBOL_GROUPS.items()
221 }
222 methods = {
223 group: inspect_methods(module_name, class_name, method_names)
224 for group, (module_name, class_name, method_names) in METHOD_GROUPS.items()
225 }
227 try:
228 dxpy = importlib.import_module("dxpy")
229 legacy_open_file = hasattr(dxpy.DXFile, "open_file")
230 except Exception:
231 legacy_open_file = None
233 return {
234 "schema_version": "1.0",
235 "python": platform.python_version(),
236 "platform": platform.platform(),
237 "documented_baseline": DOCUMENTED_BASELINE,
238 "dxpy_version": dxpy_version,
239 "version_meets_baseline": (
240 numeric_version(dxpy_version) >= numeric_version(DOCUMENTED_BASELINE)
241 ),
242 "symbols": symbols,
243 "methods": methods,
244 "known_legacy_checks": {
245 "DXFile.open_file_available": legacy_open_file,
246 "recommended_file_open": "dxpy.open_dxfile",
247 },
248 }
251def missing_required_symbols(report: dict[str, Any]) -> list[str]:
252 observed = {
253 entry["qualified_name"]: entry["available"]
254 for entries in report.get("symbols", {}).values()
255 for entry in entries
256 }
257 return sorted(name for name in REQUIRED_SYMBOLS if not observed.get(name, False))
260def missing_required_methods(report: dict[str, Any]) -> list[str]:
261 required = {
262 "DXFile.describe",
263 "DXFile.clone",
264 "DXApplet.run",
265 "DXWorkflow.run",
266 "DXJob.wait_on_done",
267 "DXJob.get_output_ref",
268 "DXAnalysis.wait_on_done",
269 "DXProject.list_folder",
270 }
271 missing = []
272 for qualified_name in sorted(required):
273 class_name, method_name = qualified_name.split(".", 1)
274 methods = report.get("methods", {}).get(class_name, {})
275 if not methods.get(method_name, {}).get("available", False):
276 missing.append(qualified_name)
277 return missing
280def print_text(report: dict[str, Any]) -> None:
281 print(f"Python: {report['python']}")
282 print(f"Platform: {report['platform']}")
283 print(f"dxpy: {report['dxpy_version'] or 'not installed'}")
284 print(f"Documented baseline: {report['documented_baseline']}")
285 print(f"Meets baseline: {report['version_meets_baseline']}")
287 for group, entries in report.get("symbols", {}).items():
288 print(f"\n[{group}]")
289 for entry in entries:
290 status = "ok" if entry["available"] else "missing"
291 line = f"- {status}: {entry['qualified_name']}"
292 if entry.get("signature"):
293 line += entry["signature"]
294 if entry.get("error"):
295 line += f" ({entry['error']})"
296 print(line)
298 print("\n[methods]")
299 for class_name, methods in report.get("methods", {}).items():
300 print(f"- {class_name}")
301 if "error" in methods:
302 print(f" error: {methods['error']}")
303 continue
304 for method_name, details in methods.items():
305 status = "ok" if details["available"] else "missing"
306 signature = details["signature"] or ""
307 print(f" - {status}: {method_name}{signature}")
309 print("\n[legacy checks]")
310 for key, value in report.get("known_legacy_checks", {}).items():
311 print(f"- {key}: {value}")
313 missing_symbols = missing_required_symbols(report)
314 missing_methods = missing_required_methods(report)
315 if missing_symbols:
316 print("\nMissing required symbols:")
317 for name in missing_symbols:
318 print(f"- {name}")
319 if missing_methods:
320 print("\nMissing required methods:")
321 for name in missing_methods:
322 print(f"- {name}")
325def build_parser() -> argparse.ArgumentParser:
326 parser = argparse.ArgumentParser(
327 description=__doc__,
328 formatter_class=argparse.ArgumentDefaultsHelpFormatter,
329 )
330 parser.add_argument(
331 "--json",
332 action="store_true",
333 help="emit machine-readable JSON",
334 )
335 parser.add_argument(
336 "--strict",
337 action="store_true",
338 help="exit non-zero if baseline symbols or methods are unavailable",
339 )
340 return parser
343def main(argv: Optional[list[str]] = None) -> int:
344 args = build_parser().parse_args(argv)
345 report = build_report()
347 if args.json:
348 print(json.dumps(report, indent=2, sort_keys=True))
349 else:
350 print_text(report)
352 if report["dxpy_version"] is None:
353 return 2
355 if args.strict:
356 if not report["version_meets_baseline"]:
357 return 1
358 if missing_required_symbols(report):
359 return 1
360 if missing_required_methods(report):
361 return 1
362 return 0
365if __name__ == "__main__":
366 sys.exit(main())