()
| 25 | |
| 26 | |
| 27 | def main(): |
| 28 | if len(sys.argv) != 3: |
| 29 | _error(f"Expected 2 arguments (module, attribute), got {len(sys.argv) - 1}") |
| 30 | |
| 31 | module_name = sys.argv[1] |
| 32 | attr_name = sys.argv[2] |
| 33 | |
| 34 | try: |
| 35 | mod = importlib.import_module(module_name) |
| 36 | except ImportError as exc: |
| 37 | _error(f"Failed to import module '{module_name}': {exc}") |
| 38 | |
| 39 | obj = getattr(mod, attr_name, None) |
| 40 | if obj is None: |
| 41 | _error(f"Module '{module_name}' has no attribute '{attr_name}'") |
| 42 | |
| 43 | fn = getattr(obj, "get_crons", None) |
| 44 | if fn is None: |
| 45 | _error( |
| 46 | f"'{module_name}.{attr_name}' has no 'get_crons' method. " |
| 47 | f"The cron entrypoint object must define a get_crons() method that " |
| 48 | f"returns a list of (module:function, schedule) pairs." |
| 49 | ) |
| 50 | |
| 51 | if not callable(fn): |
| 52 | _error(f"'{module_name}.{attr_name}.get_crons' is not callable") |
| 53 | |
| 54 | try: |
| 55 | result = fn() |
| 56 | entries = [] |
| 57 | for item in result: |
| 58 | if not (isinstance(item, (list, tuple)) and len(item) == 2): |
| 59 | _error(f"Each cron entry must be a (module:function, schedule) pair, got: {item!r}") |
| 60 | module_function, schedule = item |
| 61 | entries.append({"module_function": str(module_function), "schedule": str(schedule)}) |
| 62 | print(json.dumps({"entries": entries})) |
| 63 | except Exception as exc: |
| 64 | _error(f"Error calling '{module_name}.{attr_name}.get_crons()': {exc}") |
| 65 | |
| 66 | |
| 67 | main() |
no test coverage detected