Python contextlib 模块，ExitStack() 实例源码

我们从Python开源项目中，提取了以下46个代码示例，用于说明如何使用contextlib.ExitStack()。

项目：allennlp 作者：allenai | 项目源码 | 文件源码

def _predict(predictors: Dict[str, str]):
    def predict_inner(args: argparse.Namespace) -> None:
        predictor = _get_predictor(args, predictors)
        output_file = None

        if args.silent and not args.output_file:
            print("--silent specified without --output-file.")
            print("Exiting early because no output will be created.")
            sys.exit(0)

        # ExitStack allows us to conditionally context-manage `output_file`, which may or may not exist
        with ExitStack() as stack:
            input_file = stack.enter_context(args.input_file)  # type: ignore
            if args.output_file:
                output_file = stack.enter_context(args.output_file)  # type: ignore

            _run(predictor, input_file, output_file, args.batch_size, not args.silent, args.cuda_device)

    return predict_inner

项目：taskcv-2017-public 作者：VisionLearningGroup | 项目源码 | 文件源码

def adversarial_discriminator(net, layers, scope='adversary', leaky=False):
    if leaky:
        activation_fn = tflearn.activations.leaky_relu
    else:
        activation_fn = tf.nn.relu
    with ExitStack() as stack:
        stack.enter_context(tf.variable_scope(scope))
        stack.enter_context(
            slim.arg_scope(
                [slim.fully_connected],
                activation_fn=activation_fn,
                weights_regularizer=slim.l2_regularizer(2.5e-5)))
        for dim in layers:
            net = slim.fully_connected(net, dim)
        net = slim.fully_connected(net, 2, activation_fn=None)
    return net

项目：igd-exporter 作者：yrro | 项目源码 | 文件源码

def search(timeout):
    '''
    Search for devices implementing WANCommonInterfaceConfig on the network.

    Search ends the specified number of seconds after the last result (if any) was received.

    Returns an iterator of root device URLs.
    '''
    with contextlib.ExitStack() as stack:
        sockets = []
        sockets.append(stack.enter_context(socket.socket(socket.AF_INET6, socket.SOCK_DGRAM, socket.IPPROTO_UDP)))
        sockets.append(stack.enter_context(socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)))

        for s in sockets:
            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            if s.family == socket.AF_INET6:
                s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)

            with concurrent.futures.ThreadPoolExecutor(len(sockets)) as ex:
                return itertools.chain.from_iterable(ex.map(lambda s: search_socket(s, timeout, ns['i']), sockets))

项目：senti 作者：stevenxxiu | 项目源码 | 文件源码

def write_split_emote(cls):
        retweet_re = re.compile(r'RT\s*"?[@?][a-zA-Z0-9_]+:?')
        with open('emote/all.txt', encoding='utf-8') as in_sr, ExitStack() as stack:
            out_srs = {
                name: stack.enter_context(open('emote/class_{}.txt'.format(name), 'w', encoding='utf-8'))
                for name in ['pos', 'neg']
            }
            for i, line in enumerate(in_sr):
                if retweet_re.search(line):
                    continue
                counts = [0, 0, 0]
                for match in cls.emoticon_re.finditer(line):
                    counts[Emoticons.assess_match(match)] += 1
                label = None
                if counts[0] > 0 and counts[1] == 0 and counts[2] == 0:
                    label = 0
                elif counts[0] == 0 and counts[1] == 0 and counts[2] > 0:
                    label = 2
                if label is not None:
                    out_srs[label].write(cls.emoticon_re.sub(' ', line).strip() + '\n')

项目：MUBench 作者：stg-tud | 项目源码 | 文件源码

def post(url: str, data: object, file_paths: List[str] = None, username: str="", password: str=""):
    request = {
        "url": url,
        "data": json.dumps(data, cls=JSONFloatEncoder)
    }

    if username:
        if not password:
            password = getpass.getpass("Enter password for '{}': ".format(username))
        request["auth"] = (username, password)

    with ExitStack() as es:
        if file_paths:
            request["data"] = {"data": request["data"]}
            request["files"] = [__create_file_tuple(path, es) for path in file_paths]

        response = requests.post(**request)
        response.raise_for_status()

项目：aiotk 作者：AndreLouisCaron | 项目源码 | 文件源码

def test_exit_stack_exception_propagate():

    h1 = mock.MagicMock()
    h2 = mock.MagicMock()
    v1 = mock.MagicMock()
    v2 = mock.MagicMock()
    error = ValueError('FUUU')

    with pytest.raises(ValueError) as exc:
        with ExitStack() as stack:
            v = stack.enter_context(AutoClose(h1, v=v1))
            assert v is v1
            v = stack.enter_context(AutoClose(h2, v=v2))
            assert v is v2
            raise error
    assert exc.value is error

    h2.close.assert_called_once_with(ValueError, error, mock.ANY)
    h1.close.assert_called_once_with(ValueError, error, mock.ANY)

项目：rankedftw 作者：andersroos | 项目源码 | 文件源码

def __call__(self):
        status = 1
        with log_exception(status=1):
            args = self.parser.parse_args()
            log_args(args)
            config.log_cached()
            logger = getLogger('django')

            with ExitStack() as stack:
                if self._pid_file:
                    stack.enter_context(pid_file(dirname=config.PID_DIR, max_age=self._pid_file_max_age))

                if self._stoppable:
                    self._stoppable_instance = stoppable()
                    stack.enter_context(self._stoppable_instance)

                status = self.run(args, logger) or 0
        sys.exit(status)