一只萌萌小番薯
我最近遇到了这个问题。由于Python上游的原因,这似乎是一个难题:posix_spawn没有提供读取生成进程的环境变量的方法,也没有任何简单的方法来读取正在运行的进程的环境。Bashsource专门用于在 bash 解释器中运行 bash 代码:它只是在当前 bash 解释器中评估文件,而不是启动子进程。如果您从 Python 运行 bash 代码,则此机制无法工作。可以创建一个专门用于从 Python 运行 bash 代码的单独机制。以下是我能做到的最好的。如果有一个不那么脆弱的解决方案就好了。import jsonimport osimport subprocessimport sysfrom contextlib import AbstractContextManagerclass 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)例子: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"