Feat: Added CLI flags to start, stop, restart and check status of productions

src/intersystems_pyprod/_parser.py

@@ -9,7 +9,7 @@
 import importlib
 import importlib.util
 from operator import attrgetter
-
+from ._prod_controls import get_prod_status, start_prod, stop_prod, restart_prod
 from ._method_stubs import STUBS
 
 
@@ -854,7 +854,28 @@ def generate_msg_wrappers(tree, script_name, folder_name, iris_package_name, pyt
 
 # _______________________________________________________________________________________________________________________ generate_msg_wrappers #
 
-
+def production_control(args):
+    """
+    Handle production control commands based on the provided arguments.
+    """
+    if args.restart:
+        print("Restarting production...")
+        output = restart_prod(args.timeout, args.force)
+        print(output["message"])
+    elif args.stop:
+        print("Stopping production...")
+        output = stop_prod(args.timeout, args.force)
+        print(output["message"])
+    elif args.start:
+        print(f"Starting production {args.start}...")
+        output = start_prod(args.start)
+        print(output["message"])
+    elif args.status:
+        output = get_prod_status()
+        if output["status"]==2: 
+            print(f"No productions are currently running in {os.environ.get('IRISNAMESPACE', 'the current namespace')}.")
+        else:
+            print(f"Production {output['name']} is {output['status_message']}")
 
 def get_package_root(module_name: str) -> Path:
     spec = importlib.util.find_spec(module_name)
@@ -877,8 +898,22 @@ def main(argv: list[str] = None):
     parser.add_argument("-m", "--module", help="Dotted module to analyze (e.g. pkg.sub.mod). If set, ignore positional file.")
     parser.add_argument("-s", "--source-root", help="Project source root used to compute absolute module names when loading from a file", dest="sourceroot")
     parser.add_argument("input_script", nargs="?", help="Path to a .py file (used when -m/--module is not provided)")
+
+
+    # Production control options: 
+    group = parser.add_mutually_exclusive_group()
+    group.add_argument("-r", "--restart", action="store_true", help="Restart the production if it's already running")
+    group.add_argument("--start", metavar="PRODUCTION", help="Start the named production")
+    group.add_argument("--stop", action="store_true", help="Stop the production if it's running")
+    group.add_argument("--status", action="store_true", help="Get the status of the production")
+    parser.add_argument("--timeout", type=int, default=10, help="Timeout in seconds for stopping/restarting the production (use with --stop or --restart)")
+    parser.add_argument("--force", action="store_true", help="Force stop the production without waiting for graceful shutdown (use with --stop or --restart)")
+
+
     args = parser.parse_args(argv)
 
+    prod_control_args = any([args.restart, args.stop, args.start, args.status])
+
     # Load source and module under a correct package context
     if args.sourceroot:
         sys.path.insert(0, os.path.abspath(args.sourceroot))
@@ -909,6 +944,10 @@ def main(argv: list[str] = None):
     else:
         # File mode
         if not args.input_script:
+            # Allow running production controls without providing a script
+            if prod_control_args:
+                production_control(args)
+                sys.exit(0)
             print("You must provide either -m/--module or a path to a .py file.")
             sys.exit(2)
         try:
@@ -1006,7 +1045,8 @@ def main(argv: list[str] = None):
         loaded_module,
         args.module
     )
-
+    if prod_control_args:
+        production_control(args)
 
 
 if __name__ == "__main__":

src/intersystems_pyprod/_prod_controls.py

@@ -0,0 +1,100 @@
+import iris
+
+RUNNING = 1
+STOPPED = 2
+TROUBLED = 3
+SUSPENDED = 4
+
+
+STATUS_MESSAGES = {
+    RUNNING: "Running",
+    STOPPED: "Stopped",
+    TROUBLED: "Troubled",
+    SUSPENDED: "Suspended",
+}
+
+
+class ProductionStatusError(RuntimeError):
+    pass
+
+def get_prod_status():
+    """
+    Returns the name and status of the production as a dictionary.
+    Status codes are: 
+    1 - Running
+    2 - Stopped
+    3 - Troubled
+    4 - Suspended
+    """
+
+    production_name = iris.ref("")
+    production_status = iris.ref("")
+
+    response = iris.Ens.Director.GetProductionStatus(production_name, production_status)
+    if response != 1:
+        raise ProductionStatusError(f"Failed to get production status: {response}")
+    return {"name": production_name.value, "status": production_status.value, "status_message": STATUS_MESSAGES.get(production_status.value, "Unknown")}
+
+
+def start_prod(production_name):
+    """Starts the production."""
+
+    if not production_name or not production_name.strip():
+        return {"ok": False, "message": "Production name is required to start a production."}
+
+    prod = get_prod_status()
+    if prod['status'] in [RUNNING, TROUBLED, SUSPENDED]: # Running, Troubled, or Suspended
+            if prod['name'] == production_name:
+                return {"ok": False, "message": f"Production {prod['name']} is already active with status {prod['status_message']}. Use -r flag to restart the production."}
+            else:
+                return {"ok": False, "message": f"Production {prod['name']} is already active with status {prod['status_message']}. Please stop (--stop) the production before starting a new one."}
+
+    response = iris.Ens.Director.StartProduction(production_name)
+    if response == 1:
+            return {"ok": True, "message": f"Production {production_name} started successfully."}
+    else:
+        return {"ok": False, "message": f"Failed to start production {production_name}: {response}."}
+
+
+
+def stop_prod(timeout=10, force=False):
+    """Stops the production."""
+
+
+    if timeout < 0:
+        return {"ok": False, "message": "Timeout must be non-negative."}
+
+    prod = get_prod_status()
+
+    # Production status code 2 is $$$eProductionStateStopped
+    if prod['status'] == STOPPED:
+            return {"ok": False, "message": "There is no production running in the current namespace. Use --start flag to start a production."}
+
+
+    response = iris.Ens.Director.StopProduction(timeout, force)
+    if response == 1:
+        return {"ok": True, "message": f"Production {prod['name']} stopped successfully."}
+    else:
+        return {"ok": False, "message": f"Failed to stop production {prod['name']}: {response}. Try increasing the timeout or setting force to 1."}
+
+
+
+def restart_prod(timeout=10, force=False):
+    """Restarts the production."""
+    # iris.Ens.Director internally checks if a production is running before restarting it
+
+    if timeout < 0:
+        return {"ok": False, "message": "Timeout must be non-negative."}
+
+    prod = get_prod_status()
+    if prod['status'] == STOPPED:
+            return {"ok": False, "message": "There is no production running to restart in the current namespace. Use --start flag to start a named production."}
+
+    response = iris.Ens.Director.RestartProduction(timeout, force)
+
+    if response == 1:
+        return {"ok": True, "message": "Production restarted successfully."}
+    else:   
+        return {"ok": False, "message": f"Failed to restart production: {response}, try increasing the timeout or setting force to 1."}
+
+