python类suppress()的实例源码

def ensure_latest_consul_release(host=None, port=22):
    """
    Make sure the latest consul release is downloaded on the remote machine
    :param str host: hostname or ip of the remote machine, or None for the local machine
    :param int port: port to use to connect to the remote machine over ssh
    :return None:
    """
    log.info("Ensuring consul release {} is on disk "
             "on the remote machine".format(CONSUL_RELEASE.split('/')[-1]))
    with suppress(FileNotFoundError):
        # remove any previously existing zip in case we tried
        # before but the zip was corrupted
        remove(CONSUL_RELEASE.split('/')[-1])
    wget(
        CONSUL_RELEASE, host=host, port=port,
        failure_message="Failed to retrieve {}".format(
            CONSUL_RELEASE.split('/')[-1]
        )
    )

def run_app(app, port, loop):
    handler = app.make_handler(access_log=None, loop=loop)

    loop.run_until_complete(app.startup())
    server = loop.run_until_complete(loop.create_server(handler, HOST, port))

    try:
        loop.run_forever()
    except KeyboardInterrupt:  # pragma: no branch
        pass
    finally:
        logger.info('shutting down server...')
        server.close()
        loop.run_until_complete(server.wait_closed())
        loop.run_until_complete(app.shutdown())
        with contextlib.suppress(asyncio.TimeoutError, KeyboardInterrupt):
            loop.run_until_complete(handler.shutdown(0.1))
        with contextlib.suppress(asyncio.TimeoutError, KeyboardInterrupt):
            loop.run_until_complete(app.cleanup())
        with contextlib.suppress(KeyboardInterrupt):
            loop.close()

def loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(None)
    executor = concurrent.futures.ThreadPoolExecutor()
    loop.set_default_executor(executor)
    yield loop

    if loop.is_closed:
        pending = asyncio.Task.all_tasks(loop=loop)
        for task in pending:
            task.cancel()
            # Now we should await task to execute it's cancellation.
            # Cancelled task raises asyncio.CancelledError that we can suppress
            with suppress(asyncio.CancelledError):
                loop.run_until_complete(task)
        executor.shutdown()
        loop.close()

    gc.collect()
    asyncio.set_event_loop(None)

def apply(self):
        """Creates or deletes the autostart file based upon current state"""
        if not self.props.sensitive: # We never got the current state
            return

        if self.props.active and not self._was_enabled:
            logging.info('Creating autostart file')
            source = Gio.File.new_for_uri('resource:///se/tingping/Trg/se.tingping.Trg.service.desktop')
            if hasattr(source, 'copy_async'):
                # TODO: Fix upstream in GLib
                source.copy_async(self.autostart_file, Gio.FileCopyFlags.NONE, GLib.PRIORITY_DEFAULT)
            else:
                with suppress(GLib.Error):
                    source.copy(self.autostart_file, Gio.FileCopyFlags.NONE)
        elif not self.props.active and self._was_enabled:
            logging.info('Deleting autostart file')
            self.autostart_file.delete_async(GLib.PRIORITY_DEFAULT)

def data_mapper(mapper: Callable[[types.DataType], types.DataType]) -> Callable[[types.ListenerType], types.ListenerType]:
    '''
    Apply a mapper to the listener's `data` argument.
    '''

    def decorator(listener: types.ListenerType) -> types.ListenerType:

        @functools.wraps(listener)
        def wrapped(dtype: types.DiffType, index: int, data: types.DataType) -> None:
            listener(dtype, index, mapper(data))

        with contextlib.suppress(AttributeError):
            wrapped.finalize_batch = listener.finalize_batch

        return wrapped
    return decorator

def insert_reviews(neighborhood, db):
    listings_cursor = db['listings_' +
                         neighborhood].find({"reviews_count": {"$gt": 0}})
    listings = [listing for listing in listings_cursor]
    collection = db['reviews_' + neighborhood]
    for listing in listings:
        reviews_cursor = db.reviews_en.find({'listing_id':listing['_id']})
        reviews = [review for review in reviews_cursor]
        if reviews:
            with suppress(Exception):
                collection.insert_many(reviews, ordered=True)
            # try:
            #     collection.insert_many(reviews, ordered=True)
            # except:
            #     print(reviews)
            #     raise

def remove(self, node):
        self._nodes.remove(node)

        for n in iter(self._nodes):
            with contextlib.suppress(KeyError):
                n.edges.remove(node)

def _get_name_from_func(func, other):
    if not isinstance(func, types.LambdaType):
        with contextlib.suppress(AttributeError):
            return func.__qualname__

        with contextlib.suppress(AttributeError):
            return func.__name__

    return other

def resolve_self_references(self, rules):
        """Resolves `$self` references to actual application name in security group rules."""
        with suppress(KeyError):
            rule = rules.pop('$self')
            rules[self.app_name] = rule
        return rules

def __getattr__(self, name):
        if name != '__setstate__':
            # this avoids an infinite loop when pickle looks for the
            # __setstate__ attribute before the object is initialized
            with suppress(KeyError):
                return self._mapping[name]
        raise AttributeError("%r object has no attribute %r" %
                             (type(self).__name__, name))

def remove_project(self):
        self.close_project()
        with contextlib.suppress(FileNotFoundError):
            self.project_path.unlink()

def cleanup(ctx):
    pv = Povray(ctx.config.povray.parent_dir)
    with contextlib.suppress(FileNotFoundError):
        pv.shelve_db_path.unlink()
        pv.scitools_udb_path.unlink()

def byro_information(request):
    ctx = {'config': Configuration.get_solo()}

    try:
        ctx['url_name'] = resolve(request.path_info).url_name
    except Http404:
        ctx['url_name'] = ''

    if settings.DEBUG:
        ctx['development_warning'] = True
        with suppress(Exception):
            import subprocess
            ctx['byro_version'] = subprocess.check_output(['git', 'describe', '--always'])

    return ctx

def delete(self, sender):
    if self.control.filename != '':
      (name, _) = os.path.splitext(os.path.basename(self.control.filename))
      try:
        console.alert('Delete', name, button1='Ok')
      except KeyboardInterrupt:
        return
      os.remove(self.control.filename)
      with contextlib.suppress(FileNotFoundError):
        os.remove(self.control.filename[:-4]+'.json')
      self.control.set_default_map()
    else:
      self.control.quit_map(sender)

def match(v1, v2, nomatch=-1, incomparables=None, start=0):
    """
    Return a vector of the positions of (first)
    matches of its first argument in its second.

    Parameters
    ----------
    v1: array-like
        the values to be matched

    v2: array-like
        the values to be matched against

    nomatch: int
        the value to be returned in the case when
        no match is found.

    incomparables: array-like
        a list of values that cannot be matched.
        Any value in v1 matching a value in this list
        is assigned the nomatch value.
    start: int
        type of indexing to use. Most likely 0 or 1
    """
    lookup = {}
    for i, x in enumerate(v2):
        if x not in lookup:
            lookup[x] = i

    lst = [nomatch] * len(v1)
    skip = set(incomparables) if incomparables else set()
    for i, x in enumerate(v1):
        if x in skip:
            continue

        with suppress(KeyError):
            lst[i] = lookup[x] + start

    return lst

def suppress(*exceptions):
        """
        Return a context manager that suppresses any of the
        specified exceptions if they occur in the body of a
        with statement and then resumes execution with the
        first statement following the end of the with statement.
        """
        try:
            yield
        except exceptions:
            pass

def get_valid_kwargs(func, potential_kwargs):
    """
    Return valid kwargs to function func
    """
    kwargs = {}
    for name in get_kwarg_names(func):
        with suppress(KeyError):
            kwargs[name] = potential_kwargs[name]
    return kwargs

def init_dbs(sqls):
    # graceful initialization tries to create new tables as a test to see if this is a new DB or not
    for s in sqls:
        with suppress(sqlite3.OperationalError):
            cursor.execute(s)

def __setitem__(self, key, val):
        """Allow using [] syntax to save a keyvalue.

        - It is case-insensitive, so it will overwrite a key which only
          differs by case.
        """
        if isinstance(val, bool):
            val = '1' if val else '0'
        key_fold = key.casefold()
        for k in self.keys:
            if k.casefold() == key_fold:
                # Check case-insensitively for this key first
                orig_val = self.keys.get(k)
                self.keys[k] = str(val)
                break
        else:
            orig_val = self.keys.get(key)
            self.keys[key] = str(val)

        # Update the by_class/target dicts with our new value
        if key_fold == 'classname':
            with suppress(KeyError):
                self.map.by_class[orig_val].remove(self)
            self.map.by_class[val].add(self)
        elif key_fold == 'targetname':
            with suppress(KeyError):
                self.map.by_target[orig_val].remove(self)
            self.map.by_target[val].add(self)

def ignore_exceptions_2():
    data = None
    import contextlib
    with contextlib.suppress(FileNotFoundError, PermissionError):
        with open(DATA_FILE) as f:
            data = f.read()
    return data