update gaiaflow

Author: b-yogesh

src/gaiaflow/cli/commands/minikube.py

@@ -99,16 +99,10 @@ def restart(
     )
 
 
-@app.command(help="Containerize your package into a docker image.")
+@app.command(help="Containerize your package into a docker image inside the "
+                  "minikube cluster.")
 def dockerize(
     project_path: Path = typer.Option(..., "--path", "-p", help="Path to your project"),
-    local: bool = typer.Option(
-        False,
-        "--local",
-        "-l",
-        help="This will create the image in your local docker instance instead "
-        "of creating it inside prod-local environment",
-    ),
 ):
     imports = load_imports()
     gaiaflow_path, user_project_path = imports.create_gaiaflow_context_path(
@@ -122,7 +116,7 @@ def dockerize(
         gaiaflow_path=gaiaflow_path,
         user_project_path=user_project_path,
         action=imports.ExtendedAction.DOCKERIZE,
-        local=local,
+        local=False,
     )
 
 

src/gaiaflow/cli/commands/mlops.py

@@ -12,15 +12,18 @@
 
 
 def load_imports():
-    from gaiaflow.constants import BaseAction
+    from gaiaflow.constants import BaseAction, ExtendedAction
     from gaiaflow.managers.mlops_manager import MlopsManager
+    from gaiaflow.managers.minikube_manager import MinikubeManager
     from gaiaflow.managers.utils import (create_gaiaflow_context_path,
                                          gaiaflow_path_exists_in_state,
                                          save_project_state)
 
     return SimpleNamespace(
         BaseAction=BaseAction,
+        ExtendedAction=ExtendedAction,
         MlopsManager=MlopsManager,
+        MinikubeManager=MinikubeManager,
         create_gaiaflow_context_path=create_gaiaflow_context_path,
         gaiaflow_path_exists_in_state=gaiaflow_path_exists_in_state,
         save_project_state=save_project_state,
@@ -233,6 +236,34 @@ def cleanup(
     )
 
 
+
+@app.command(help="Containerize your package into a docker image locally.")
+def dockerize(
+    project_path: Path = typer.Option(..., "--path", "-p", help="Path to your project"),
+):
+    imports = load_imports()
+    gaiaflow_path, user_project_path = imports.create_gaiaflow_context_path(
+        project_path
+    )
+    gaiaflow_path_exists = imports.gaiaflow_path_exists_in_state(gaiaflow_path, True)
+    if not gaiaflow_path_exists:
+        imports.save_project_state(user_project_path, gaiaflow_path)
+    else:
+        typer.echo(
+            f"Gaiaflow project already exists at {gaiaflow_path}. Skipping "
+            f"saving to the state"
+        )
+
+    typer.echo("Running dockerize")
+    imports.MinikubeManager.run(
+        gaiaflow_path=gaiaflow_path,
+        user_project_path=user_project_path,
+        action=imports.ExtendedAction.DOCKERIZE,
+        local=True,
+    )
+
+
+
 # TODO: To let the user update the current infra with new local packages or
 #  mounts as they want it.
 # def update():

src/gaiaflow/core/create_task.py

@@ -309,15 +309,15 @@
 
 class GaiaflowMode(Enum):
     DEV = "dev"
-    PROD_LOCAL_DOCKER = "prod_local_docker"
-    PROD_LOCAL_MINIKUBE = "prod_local_minikube"
+    DEV_DOCKER = "dev_docker"
+    PROD_LOCAL = "prod_local"
     PROD = "prod"
 
 
 OPERATOR_MAP = {
     GaiaflowMode.DEV: DevTaskOperator,
-    GaiaflowMode.PROD_LOCAL_DOCKER: DockerTaskOperator,
-    GaiaflowMode.PROD_LOCAL_MINIKUBE: ProdLocalTaskOperator,
+    GaiaflowMode.DEV_DOCKER: DockerTaskOperator,
+    GaiaflowMode.PROD_LOCAL: ProdLocalTaskOperator,
     GaiaflowMode.PROD: ProdTaskOperator,
 }
 
@@ -333,8 +333,12 @@ def create_task(
     env_vars: dict | None = None,
     retries: int = 3,
     dag=None,
-    params=None,
 ):
+    """It is a high-level abstraction on top of Apache Airflow operators.
+
+    It allows you to define tasks for your DAGs in a uniform, environment-aware
+    way, without worrying about which Airflow operator to use for each execution mode.
+    """
     try:
         gaiaflow_mode: GaiaflowMode = GaiaflowMode(mode)
     except ValueError:
@@ -348,7 +352,6 @@ def create_task(
         env_vars = {}
 
     dag_params = getattr(dag, "params", {}) if dag else {}
-    combined_params = {**dag_params, **(params or {})}
 
     operator_cls = OPERATOR_MAP.get(gaiaflow_mode)
     if not operator_cls:
@@ -363,7 +366,7 @@ def create_task(
         secrets=secrets,
         env_vars=env_vars,
         retries=retries,
-        params=combined_params,
+        params=dag_params,
         mode=mode,
     )