I ran into this issue just recently. It seems that this is a difficult problem for reasons upstream of Python: posix_spawn
doesn't give a way to read the environment variables of the spawned process, nor is there any easy way to read the environment of a running process.
Bash's source
is specific to running bash code in the bash interpreter: it just evals the file in the current bash interpreter rather than starting a subprocess. This mechanism can't work if you are running bash code from Python.
It is possible to make a separate mechanism specific to running bash code from Python. The following is the best that I could manage. Would be nice to have a less flimsy solution.
import json
import os
import subprocess
import sys
from contextlib import AbstractContextManager
class BashRunnerWithSharedEnvironment(AbstractContextManager):
"""Run multiple bash scripts with persisent environment.
Environment is stored to "env" member between runs. This can be updated
directly to adjust the environment, or read to get variables.
"""
def __init__(self, env=None):
if env is None:
env = dict(os.environ)
self.env: Dict[str, str] = env
self._fd_read, self._fd_write = os.pipe()
def run(self, cmd, **opts):
if self._fd_read is None:
raise RuntimeError("BashRunner is already closed")
write_env_pycode = ";".join(
[
"import os",
"import json",
f"os.write({self._fd_write}, json.dumps(dict(os.environ)).encode())",
]
)
write_env_shell_cmd = f"{sys.executable} -c '{write_env_pycode}'"
cmd += "\n" + write_env_shell_cmd
result = subprocess.run(
["bash", "-ce", cmd], pass_fds=[self._fd_write], env=self.env, **opts
)
self.env = json.loads(os.read(self._fd_read, 5000).decode())
return result
def __exit__(self, exc_type, exc_value, traceback):
if self._fd_read:
os.close(self._fd_read)
os.close(self._fd_write)
self._fd_read = None
self._fd_write = None
def __del__(self):
self.__exit__(None, None, None)
Example:
with BashRunnerWithSharedEnvironment() as bash_runner:
bash_runner.env.pop("A", None)
res = bash_runner.run("A=6; echo $A", stdout=subprocess.PIPE)
assert res.stdout == b'6\n'
assert bash_runner.env.get("A", None) is None
bash_runner.run("export A=2")
assert bash_runner.env["A"] == "2"
res = bash_runner.run("echo $A", stdout=subprocess.PIPE)
assert res.stdout == b'2\n'
res = bash_runner.run("A=6; echo $A", stdout=subprocess.PIPE)
assert res.stdout == b'6\n'
assert bash_runner.env.get("A", None) == "6"
bash_runner.env["A"] = "7"
res = bash_runner.run("echo $A", stdout=subprocess.PIPE)
assert res.stdout == b'7\n'
assert bash_runner.env["A"] == "7"