Add API option to specify channels

mamba_gator/envmanager.py

@@ -340,18 +340,25 @@ async def conda_config(self) -> Dict[str, Any]:
         _, output = ans
         return self._clean_conda_json(output)
 
-    async def clone_env(self, env: str, name: str) -> Dict[str, str]:
+    async def clone_env(self, env: str, name: str, channels: Optional[List[str]] = None) -> Dict[str, str]:
         """Clone an environment.
 
         Args:
             env (str): To-be-cloned environment name
             name (str): New environment name
+            channels (List[str], optional): Channel priority list (e.g., ["conda-forge", "defaults"])
 
         Returns:
             Dict[str, str]: Clone command output.
         """
+        cmd_args = ["-y", "-q", "--json", "-n", name]
+
+        if channels:
+            for channel in channels:
+                cmd_args.extend(["-c", channel])
+
         ans = await self._execute(
-            self.manager, "create", "-y", "-q", "--json", "-n", name, "--clone", env
+            self.manager, "create", *cmd_args, "--clone", env
         )
 
         rcode, output = ans
@@ -360,18 +367,25 @@ async def clone_env(self, env: str, name: str) -> Dict[str, str]:
 
         return self._clean_conda_json(output)
 
-    async def create_env(self, env: str, *args) -> Dict[str, str]:
+    async def create_env(self, env: str, *args, channels: Optional[List[str]] = None) -> Dict[str, str]:
         """Create a environment from a list of packages.
 
         Args:
             env (str): Name of the environment
             *args (List[str]): optional, packages to install
+            channels (List[str], optional): Channel priority list (e.g., ["conda-forge", "defaults"])
 
         Returns:
             Dict[str, str]: Create command output
         """
+        cmd_args = ["-y", "-q", "--json", "-n", env]
+
+        if channels:
+            for channel in channels:
+                cmd_args.extend(["-c", channel])
+
         ans = await self._execute(
-            self.manager, "create", "-y", "-q", "--json", "-n", env, *args
+            self.manager, "create", *cmd_args, *args
         )
 
         rcode, output = ans
@@ -565,14 +579,15 @@ def get_info(env):
         return {"environments": envs_list}
 
     async def update_env(
-        self, env: str, file_content: str, file_name: str = "environment.yml"
+        self, env: str, file_content: str, file_name: str = "environment.yml", channels: Optional[List[str]] = None
     ) -> Dict[str, str]:
         """Update a environment from a file.
 
         Args:
             env (str): Name of the environment
             file_content (str): File content
             file_name (str): optional, Original filename
+            channels (List[str], optional): Channel priority list (e.g., ["conda-forge", "defaults"])
 
         Returns:
             Dict[str, str]: Update command output
@@ -590,9 +605,14 @@ async def update_env(
         if file_name.endswith('.txt'):
             # For .txt files (explicit package lists), use conda install
             self.log.debug(f"Updating environment {env} with txt file using conda install")
-            ans = await self._execute(
-                self.manager, "install", "-y", "-q", "--json", "-n", env, "--file", name
-            )
+            cmd_args = ["-y", "-q", "--json", "-n", env]
+
+            if channels:
+                for channel in channels:
+                    cmd_args.extend(["-c", channel])
+
+            cmd_args.extend(["--file", name])
+            ans = await self._execute(self.manager, "install", *cmd_args)
         else:
             # For .yml files (environment definitions), use conda env update
             self.log.debug(f"Updating environment {env} with yml file using conda env update")
@@ -1032,19 +1052,26 @@ async def check_update(
             # no action plan returned means everything is already up to date
             return {"updates": []}
 
-    async def install_packages(self, env: str, packages: List[str]) -> Dict[str, str]:
+    async def install_packages(self, env: str, packages: List[str], channels: Optional[List[str]] = None) -> Dict[str, str]:
         """Install packages in an environment.
 
         Args:
             env (str): Environment name
             packages (List[str]): List of packages to install
+            channels (List[str], optional): Channel priority list (e.g., ["conda-forge", "defaults"])
 
         Returns:
             Dict[str, str]: Install command output.
         """
-        ans = await self._execute(
-            self.manager, "install", "-y", "-q", "--json", "-n", env, *packages
-        )
+        cmd_args = ["-y", "-q", "--json", "-n", env]
+
+        if channels:
+            for channel in channels:
+                cmd_args.extend(["-c", channel])
+
+        cmd_args.extend(packages)
+
+        ans = await self._execute(self.manager, "install", *cmd_args)
         _, output = ans
         return self._clean_conda_json(output)
 
@@ -1134,19 +1161,26 @@ async def develop_packages(
 
         return {"packages": result}
 
-    async def update_packages(self, env: str, packages: List[str]) -> Dict[str, str]:
+    async def update_packages(self, env: str, packages: List[str], channels: Optional[List[str]] = None) -> Dict[str, str]:
         """Update packages in an environment.
 
         Args:
             env (str): Environment name
             packages (List[str]): List of packages to update
+            channels (List[str], optional): Channel priority list (e.g., ["conda-forge", "defaults"])
 
         Returns:
             Dict[str, str]: Update command output.
         """
-        ans = await self._execute(
-            self.manager, "update", "-y", "-q", "--json", "-n", env, *packages
-        )
+        cmd_args = ["-y", "-q", "--json", "-n", env]
+
+        if channels:
+            for channel in channels:
+                cmd_args.extend(["-c", channel])
+
+        cmd_args.extend(packages)
+
+        ans = await self._execute(self.manager, "update", *cmd_args)
         _, output = ans
         return self._clean_conda_json(output)
 

mamba_gator/handlers.py

@@ -77,23 +77,24 @@ def get(self, idx: int) -> Any:
         else:
             return None
 
-    def put(self, task: Callable, *args) -> int:
+    def put(self, task: Callable, *args, **kwargs) -> int:
         """Add a asynchronous task into the queue.
 
         Args:
             task (Callable): Asynchronous task
             *args : arguments of the task
+            **kwargs : keyword arguments of the task
 
         Returns:
             int: Task id
         """
         ActionsStack.__last_index += 1
         idx = ActionsStack.__last_index
 
-        async def execute_task(idx, f, *args) -> Any:
+        async def execute_task(idx, f, *args, **kwargs) -> Any:
             try:
                 get_logger().debug("Will execute task {}.".format(idx))
-                result = await f(*args)
+                result = await f(*args, **kwargs)
             except asyncio.CancelledError:
                 raise
             except Exception as e:
@@ -110,7 +111,7 @@ async def execute_task(idx, f, *args) -> Any:
 
             return result
 
-        self.__tasks[idx] = asyncio.ensure_future(execute_task(idx, task, *args))
+        self.__tasks[idx] = asyncio.ensure_future(execute_task(idx, task, *args, **kwargs))
         return idx
 
     def __del__(self):
@@ -191,6 +192,7 @@ def post(self):
             twin (str): optional, environment name to clone
             file (str): optional, environment file (TXT or YAML format)
             filename (str): optional, environment filename of the `file` content
+            channels (List[str]): optional, channel priority list (e.g., ["conda-forge", "defaults"])
         }
         """
         data = self.get_json_body()
@@ -199,17 +201,18 @@ def post(self):
         twin = data.get("twin", None)
         file_content = data.get("file", None)
         file_name = data.get("filename", "environment.txt")
+        channels = data.get("channels", None)
 
         if packages is not None:
-            idx = self._stack.put(self.env_manager.create_env, name, *packages)
+            idx = self._stack.put(self.env_manager.create_env, name, *packages, channels=channels)
         elif twin is not None:
-            idx = self._stack.put(self.env_manager.clone_env, twin, name)
+            idx = self._stack.put(self.env_manager.clone_env, twin, name, channels=channels)
         elif file_content is not None:
             idx = self._stack.put(
                 self.env_manager.import_env, name, file_content, file_name
             )
         else:
-            idx = self._stack.put(self.env_manager.create_env, name)
+            idx = self._stack.put(self.env_manager.create_env, name, channels=channels)
 
         self.redirect_to_task(idx)
 
@@ -270,13 +273,15 @@ def patch(self, env: str):
         {
             file (str): optional, environment file (TXT or YAML format)
             filename (str): optional, environment filename of the `file` content
+            channels (List[str]): optional, channel priority list (e.g., ["conda-forge", "defaults"])
         }
         """
         data = self.get_json_body()
         file_content = data["file"]
         file_name = data.get("filename", "environment.yml")
+        channels = data.get("channels", None)
 
-        idx = self._stack.put(self.env_manager.update_env, env, file_content, file_name)
+        idx = self._stack.put(self.env_manager.update_env, env, file_content, file_name, channels=channels)
 
         self.redirect_to_task(idx)
 
@@ -307,11 +312,13 @@ def patch(self, env: str):
         Request json body:
         {
             packages (List[str]): optional, list of packages to update
+            channels (List[str]): optional, channel priority list (e.g., ["conda-forge", "defaults"])
         }
         """
         body = self.get_json_body() or {}
         packages = body.get("packages", ["--all"])
-        idx = self._stack.put(self.env_manager.update_packages, env, packages)
+        channels = body.get("channels", None)
+        idx = self._stack.put(self.env_manager.update_packages, env, packages, channels=channels)
         self.redirect_to_task(idx)
 
     @tornado.web.authenticated
@@ -327,16 +334,18 @@ def post(self, env: str):
         Request json body:
         {
             packages (List[str]): list of packages to install
+            channels (List[str]): optional, channel priority list (e.g., ["conda-forge", "defaults"])
         }
         """
         body = self.get_json_body()
         packages = body["packages"]
+        channels = body.get("channels", None)
         develop = int(self.get_query_argument("develop", 0))
 
         if develop:
             idx = self._stack.put(self.env_manager.develop_packages, env, packages)
         else:
-            idx = self._stack.put(self.env_manager.install_packages, env, packages)
+            idx = self._stack.put(self.env_manager.install_packages, env, packages, channels=channels)
         self.redirect_to_task(idx)