conda_subprocess demo

`conda_subprocess` demo#

conda_subprocess runs a subprocess (or a whole Python function) inside a different conda environment than the one your current Python process is using - without having to conda activate first.

It re-implements the familiar subprocess interface (call, check_call, check_output, run, Popen) and adds two new keyword arguments to all of them:

prefix_name - the name of the target conda environment (e.g. "py312")
prefix_path - the absolute path of the target conda environment (e.g. "/home/user/mambaforge/envs/py312")

In addition, it provides a @conda decorator to execute a whole Python function in another environment, built on top of executorlib.

This notebook demonstrates both, based on the project’s own test suite and docstrings.

Setup#

To make this notebook self-contained we create two small demo conda environments:

cs-demo-sub (Python 3.11) - used for the subprocess-interface examples below. It deliberately uses a different Python version than the kernel running this notebook, to show that conda_subprocess does not care which interpreter is used in the target environment.
a second environment matching this kernel’s own Python version - used for the @conda decorator examples later on. The decorator ships a function’s code object via cloudpickle to a worker process, and cloudpickle can only do that safely between matching Python minor versions - this mirrors the constraint already encoded in the project’s own test suite (tests/test_conda_function.py skips unless the test environment is Python 3.14).

Feel free to skip the creation cell if you already have suitable environments and just adjust the names below.

import os
import shutil
import subprocess
import sys

# `conda_subprocess` needs to know where the `conda` executable lives. Inside an activated conda shell this
# is normally set automatically, but in case it isn't (e.g. the Jupyter kernel was launched without sourcing
# conda's shell hooks) we fall back to discovering it on PATH, or next to the current Python interpreter -
# which is where `conda` lives if this notebook's kernel is itself a conda environment.
if "CONDA_EXE" not in os.environ:
    candidates = [
        shutil.which("conda"),
        shutil.which("mamba"),
        os.path.join(os.path.dirname(sys.executable), "conda"),
    ]
    conda_exe = next((c for c in candidates if c and os.path.isfile(c)), None)
    assert conda_exe is not None, "Could not locate a conda executable"
    os.environ["CONDA_EXE"] = conda_exe

print("Using CONDA_EXE =", os.environ["CONDA_EXE"])
print("Notebook kernel Python:", sys.version)

Using CONDA_EXE = /srv/conda/bin/conda
Notebook kernel Python: 3.13.8 | packaged by conda-forge | (main, Oct 13 2025, 14:15:33) [GCC 14.3.0]

SUB_ENV_NAME = "cs-demo-sub"
DECORATOR_ENV_NAME = f"py{sys.version_info.major}{sys.version_info.minor}"


def conda_env_exists(name):
    result = subprocess.run(
        [os.environ["CONDA_EXE"], "env", "list"],
        capture_output=True,
        text=True,
        check=True,
    )
    return any(
        line.split()[0] == name
        for line in result.stdout.splitlines()
        if line and not line.startswith("#")
    )


def create_env(name, packages):
    if conda_env_exists(name):
        print(f"Environment '{name}' already exists, skipping creation.")
        return
    print(f"Creating environment '{name}' with {packages} ...")
    subprocess.run(
        [
            os.environ["CONDA_EXE"],
            "create",
            "-n",
            name,
            "-c",
            "conda-forge",
            *packages,
            "-y",
        ],
        capture_output=True,
        text=True,
        check=True,
    )
    print(f"Environment '{name}' created.")


create_env(SUB_ENV_NAME, ["python=3.12"])
# the decorator environment additionally needs `executorlib` (matching the version installed in this kernel)
# plus its `cloudpickle`/`pyzmq` dependencies, since that's what actually executes the shipped-over function
import executorlib

create_env(
    DECORATOR_ENV_NAME,
    [
        f"python={sys.version_info.major}.{sys.version_info.minor}",
        f"executorlib={executorlib.__version__}",
    ],
)

Creating environment 'cs-demo-sub' with ['python=3.12'] ...
Environment 'cs-demo-sub' created.
Creating environment 'py313' with ['python=3.13', 'executorlib=1.10.1'] ...
Environment 'py313' created.

Subprocess interface#

conda_subprocess mirrors the standard subprocess module. Every function accepts the same arguments as its subprocess counterpart, plus prefix_name/prefix_path to select the target environment.

`check_output`#

from conda_subprocess import check_output

check_output("python --version", prefix_name=SUB_ENV_NAME)

b'Python 3.12.13\n'

Equivalently, the environment can be selected by its absolute path with prefix_path:

import subprocess as _sp

# resolve the absolute path of the environment we just created
env_path = _sp.run(
    [
        os.environ["CONDA_EXE"],
        "run",
        "-n",
        SUB_ENV_NAME,
        "python",
        "-c",
        "import sys; print(sys.prefix)",
    ],
    capture_output=True,
    text=True,
    check=True,
).stdout.strip()
print("env_path =", env_path)

check_output("python --version", prefix_path=env_path)

env_path = /srv/conda/envs/cs-demo-sub

b'Python 3.12.13\n'

The command can also be passed as a list, just like in subprocess:

check_output(["python", "--version"], prefix_name=SUB_ENV_NAME)

b'Python 3.12.13\n'

Decoding bytes to text is supported via universal_newlines/text, exactly like in subprocess:

check_output("python --version", prefix_name=SUB_ENV_NAME, universal_newlines=True)

'Python 3.12.13\n'

`call` and `check_call`#

Both just return (or raise on) the return code, output is not captured:

from conda_subprocess import call

returncode = call("python --version", prefix_name=SUB_ENV_NAME)
print("returncode:", returncode)

Python 3.12.13
returncode: 0

from conda_subprocess import check_call

check_call("python --version", prefix_name=SUB_ENV_NAME)

Python 3.12.13

`run`#

run() returns a regular subprocess.CompletedProcess object:

from conda_subprocess import run

run("python --version", prefix_name=SUB_ENV_NAME)

Python 3.12.13

CompletedProcess(args=['/usr/bin/bash', '/tmp/tmp_dzz4mn3'], returncode=0)

Use capture_output=True (or stdout=PIPE/stderr=PIPE) to capture the output instead of inheriting the parent’s streams:

completed = run(
    "python --version",
    prefix_name=SUB_ENV_NAME,
    capture_output=True,
    universal_newlines=True,
)
print("stdout:", completed.stdout)
print("returncode:", completed.returncode)

stdout: Python 3.12.13

returncode: 0

`Popen`#

For interactive communication, conda_subprocess implements the Popen interface directly:

from subprocess import PIPE
from conda_subprocess import Popen

process = Popen(["python", "--version"], stdout=PIPE, prefix_name=SUB_ENV_NAME)
stdout, stderr = process.communicate()
print(stdout, stderr)

b'Python 3.12.13\n' None

Passing environment variables#

env is merged on top of the current process environment (it is not used to replace it):

output = check_output(
    "env",
    prefix_name=SUB_ENV_NAME,
    env={"DEMO_VARIABLE": "hello from conda_subprocess"},
    universal_newlines=True,
)
[line for line in output.splitlines() if line.startswith("DEMO_VARIABLE")]

['DEMO_VARIABLE=hello from conda_subprocess']

A note on string vs. list commands#

When the command is given as a single string, conda_subprocess splits it on whitespace with plain str.split() - it does not do shell-style tokenizing (no shlex). So arguments containing spaces or quotes (e.g. a python -c "..." snippet) must be passed as a list instead, exactly like in subprocess when shell=False:

# list form: each argument is its own list element, so the `-c` snippet survives intact even though it
# contains spaces and a semicolon - naively `.split()`-ing this as a single string would break it apart
check_output(
    ["python", "-c", "print('two plus two is', 2 + 2)"],
    prefix_name=SUB_ENV_NAME,
    universal_newlines=True,
)

'two plus two is 4\n'

Error handling#

Just like subprocess, a non-zero exit code raises CalledProcessError from check_output/check_call:

from subprocess import CalledProcessError

try:
    check_output(["python", "-c", "import sys; sys.exit(1)"], prefix_name=SUB_ENV_NAME)
except CalledProcessError as error:
    print(f"Caught expected error: returncode={error.returncode}")

Caught expected error: returncode=1

Timeouts#

The timeout argument works exactly as in subprocess, raising TimeoutExpired:

from subprocess import TimeoutExpired

try:
    call(
        ["python", "-c", "import time; time.sleep(5)"],
        prefix_name=SUB_ENV_NAME,
        timeout=1,
    )
except TimeoutExpired:
    print("Caught expected TimeoutExpired")

Caught expected TimeoutExpired

The `@conda` decorator#

In addition to the subprocess interface, conda_subprocess provides a @conda decorator (built on top of executorlib) that runs an entire Python function - arguments, return value and all - inside another conda environment.

As noted in Setup above, this relies on cloudpickle to ship the function to a worker process, so the target environment’s Python version should match the version running this notebook.

from conda_subprocess.decorator import conda
from executorlib.standalone.serialize import cloudpickle_register


@conda(prefix_name=DECORATOR_ENV_NAME, hostname_localhost=True)
def add_in_other_env(parameter_1, parameter_2):
    import os

    return parameter_1 + parameter_2, os.environ["CONDA_PREFIX"]


cloudpickle_register(ind=1)
number, conda_prefix = add_in_other_env(parameter_1=1, parameter_2=2)
print("result:", number)
print("executed inside:", conda_prefix)

result: 3
executed inside: /srv/conda/envs/py313

Combining with `executorlib`#

Because the @conda decorator is built on executorlib, decorated functions can also be submitted to an executorlib executor, e.g. to run several calls concurrently:

from executorlib import SingleNodeExecutor

cloudpickle_register(ind=1)
with SingleNodeExecutor(max_cores=1, hostname_localhost=True) as exe:
    future = exe.submit(add_in_other_env, 2, 3)
    number, conda_prefix = future.result()

print("result:", number)
print("executed inside:", conda_prefix)

result: 5
executed inside: /srv/conda/envs/py313

Remarks#

The shell parameter is not supported by Popen() and all derived functions - commands are always executed via the conda activation wrapper script, not a shell.
The pipesize and process_group parameters were removed for compatibility with Python 3.9.
env is merged on top of the current environment rather than replacing it (see the example above).

Cleanup#

The cell below removes the demo environments created at the start of this notebook. It is commented out by default so re-running the notebook doesn’t unexpectedly delete your environments - uncomment if you want a clean slate.

# for name in (SUB_ENV_NAME, DECORATOR_ENV_NAME):
#     subprocess.run([os.environ["CONDA_EXE"], "env", "remove", "-n", name, "-y"], check=True)

conda_subprocess demo

Contents

conda_subprocess demo#