运行子进程后如何获取环境变量

Question

我正在使用

subprocess.call

从我正在集成的另一个应用程序执行 shell 脚本。该脚本使用

export MY_VAR=foo

设置环境变量。接下来，我需要使用 shell 脚本设置的环境在子进程上执行更多命令。

如何从子进程中提取环境状态？它仅返回

errno

代码。

即我想跑步：

subprocess.call(["export", "MY_VAR=foo"]
subprocess.call(["echo", "$MY_VAR"])  # should print 'foo'.

我知道我可以使用

env

关键字设置环境，但我的问题的重点是如何获取子进程设置的环境变量。在 shell 中，您可以

source

任何脚本来获取其声明的环境变量。 python 中的替代方案是什么？

Answer 1

我最近遇到了这个问题。由于Python上游的原因，这似乎是一个难题：

posix_spawn

没有提供读取生成进程的环境变量的方法，也没有任何简单的方法来读取正在运行的进程的环境。

Bash 的

source

特定于在 bash 解释器中运行 bash 代码：它只是在当前 bash 解释器中评估文件，而不是启动子进程。如果您从 Python 运行 bash 代码，则此机制无法工作。

可以创建一个专门用于从 Python 运行 bash 代码的单独机制。以下是我能做到的最好的。如果有一个不那么脆弱的解决方案就好了。

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)

示例：

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"

Answer 2

这是不可能的，因为环境仅在子进程中改变。您可以从那里将其作为输出返回到 STDOUT、STDERR - 但是一旦子进程终止，您就无法从中访问任何内容。

# this is process #1
subprocess.call(["export", "MY_VAR=foo"]

# this is process #2 - it can not see the environment of process #1
subprocess.call(["echo", "$MY_VAR"])  # should print 'foo'.

Answer 3

不确定我是否看到了这里的问题。您只需要记住以下几点：

• 启动的每个子流程独立于之前子流程中完成的任何设置
• 如果您想设置一些变量并使用它们，请在ONE过程中完成这两件事

---

所以像这样制作

setupVars.sh

：

export vHello="hello"
export vDate=$(date)
export vRandom=$RANDOM

然后像这样制作

printVars.sh

：

#!/bin/bash
echo $vHello, $vDate, $vRandom

并使用以下命令使其可执行：

chmod +x printVars.sh

---

现在你的 Python 看起来像这样：

import subprocess

subprocess.call(["bash","-c","source setupVars.sh; ./printVars.sh"])

输出

hello, Mon Jul 12 00:32:29 BST 2021, 8615

Answer 4

一个非常简单的解决方案：

1. 将第一个进程环境变量保存在类似

   first-process-command && env > /tmp/env.txt

   的文件中。
2. 从文件中获取环境变量并将它们传递给第二个进程。

工作示例（Python 版本 >= 3.6）：

import os
import subprocess


SHARED_ENV_FILEPATH = '/tmp/env.txt'


def run(command, env = None):
    env = env if env is not None else {}
    process = subprocess.Popen(
        f'{command} && env > {SHARED_ENV_FILEPATH}',
        env=env,
        shell=True)
    process.wait()
    return get_env()


def get_env():
    env = {}

    with open(SHARED_ENV_FILEPATH, 'r') as subprocesses_env_file:
        for line in subprocesses_env_file.readlines():
            key = line.strip().split('=')[0]
            value = line.strip().split('=')[1]
            env[key] = value
    
    os.remove(SHARED_ENV_FILEPATH)

    return env


shared_env = run('export MY_VAR="bye" && echo "hello"')
run('echo $MY_VAR', shared_env)

也许您想从

PWD

字典中删除一些环境变量，例如

OLDPWD

或

env

（如果存在）。