Python logging 模块，info() 实例源码

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

项目：kas 作者：siemens | 项目源码 | 文件源码

def kas(argv):
    """
        The main entry point of kas.
    """
    create_logger()

    parser = kas_get_argparser()
    args = parser.parse_args(argv)

    if args.debug:
        logging.getLogger().setLevel(logging.DEBUG)

    logging.info('%s %s started', os.path.basename(sys.argv[0]), __version__)

    loop = asyncio.get_event_loop()

    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig, interruption)
    atexit.register(_atexit_handler)

    for plugin in getattr(kasplugin, 'plugins', []):
        if plugin().run(args):
            return

    parser.print_help()

项目：PyPlanet 作者：PyPlanet | 项目源码 | 文件源码

def check(self, first_check=False):
        from pyplanet import __version__ as current_version

        logging.debug('Checking for new versions...')

        async with aiohttp.ClientSession() as session:
            async with session.get(self.url) as resp:
                for release in await resp.json():
                    if not release['draft'] and not release['prerelease']:
                        self.latest = release['tag_name']
                        break

                self.current = current_version
                logging.debug('Version check, your version: {}, online version: {}'.format(self.current, self.latest))

                if first_check and self.update_available:
                    logging.info('New version of PyPlanet available, consider updating: {}'.format(self.latest))
                    await self.instance.chat(
                        '\uf1e6 $FD4$oPy$369Planet$z$s$fff \uf0e7 new version available: v{}. Consider updating!'.format(self.latest)
                    )

项目：nanoQC 作者：wdecoster | 项目源码 | 文件源码

def main():
    args = get_args()
    logging.basicConfig(
        format='%(asctime)s %(message)s',
        filename=os.path.join(args.outdir, "NanoQC.log"),
        level=logging.INFO)
    logging.info("NanoQC started.")
    sizeRange = length_histogram(
        fqin=gzip.open(args.fastq, 'rt'),
        name=os.path.join(args.outdir, "SequenceLengthDistribution.png"))
    fq = get_bin(gzip.open(args.fastq, 'rt'), sizeRange)
    logging.info("Using {} reads for plotting".format(len(fq)))
    fqbin = [dat[0] for dat in fq]
    qualbin = [dat[1] for dat in fq]
    logging.info("Creating plots...")
    per_base_sequence_content_and_quality(fqbin, qualbin, args.outdir, args.format)
    logging.info("per base sequence content and quality completed.")
    logging.info("Finished!")

项目：tsproxy 作者：WPO-Foundation | 项目源码 | 文件源码

def run(self):
    global lock, background_activity_count
    try:
      logging.debug('[{0:d}] AsyncDNS - calling getaddrinfo for {1}:{2:d}'.format(self.client_id, self.hostname, self.port))
      addresses = socket.getaddrinfo(self.hostname, self.port)
      logging.info('[{0:d}] Resolving {1}:{2:d} Completed'.format(self.client_id, self.hostname, self.port))
    except:
      addresses = ()
      logging.info('[{0:d}] Resolving {1}:{2:d} Failed'.format(self.client_id, self.hostname, self.port))
    message = {'message': 'resolved', 'connection': self.client_id, 'addresses': addresses, 'localhost': self.is_localhost}
    self.result_pipe.SendMessage(message, False)
    lock.acquire()
    if background_activity_count > 0:
      background_activity_count -= 1
    lock.release()
    # open and close a local socket which will interrupt the long polling loop to process the message
    s = socket.socket()
    s.connect((server.ipaddr, server.port))
    s.close()


########################################################################################################################
#   TCP Client
########################################################################################################################

项目：tsproxy 作者：WPO-Foundation | 项目源码 | 文件源码

def HandleResolve(self, message):
    global in_pipe,  map_localhost, lock, background_activity_count
    self.did_resolve = True
    is_localhost = False
    if 'hostname' in message:
      self.hostname = message['hostname']
    self.port = 0
    if 'port' in message:
      self.port = message['port']
    logging.info('[{0:d}] Resolving {1}:{2:d}'.format(self.client_id, self.hostname, self.port))
    if self.hostname == 'localhost':
      self.hostname = '127.0.0.1'
    if self.hostname == '127.0.0.1':
      logging.info('[{0:d}] Connection to localhost detected'.format(self.client_id))
      is_localhost = True
    if (dest_addresses is not None) and (not is_localhost or map_localhost):
      logging.info('[{0:d}] Resolving {1}:{2:d} to mapped address {3}'.format(self.client_id, self.hostname, self.port, dest_addresses))
      self.SendMessage('resolved', {'addresses': dest_addresses, 'localhost': False})
    else:
      lock.acquire()
      background_activity_count += 1
      lock.release()
      self.state = self.STATE_RESOLVING
      self.dns_thread = AsyncDNS(self.client_id, self.hostname, self.port, is_localhost, in_pipe)
      self.dns_thread.start()

项目：tsproxy 作者：WPO-Foundation | 项目源码 | 文件源码

def handle_message(self, message):
    if message['message'] == 'data' and 'data' in message and len(message['data']) > 0:
      self.buffer += message['data']
      if self.state == self.STATE_CONNECTED:
        self.handle_write()
    elif message['message'] == 'resolved':
      self.HandleResolved(message)
    elif message['message'] == 'connected':
      self.HandleConnected(message)
      self.handle_write()
    elif message['message'] == 'closed':
      if len(self.buffer) == 0:
        logging.info('[{0:d}] Server connection close being processed, closing Browser connection'.format(self.client_id))
        self.handle_close()
      else:
        logging.info('[{0:d}] Server connection close being processed, queuing browser connection close'.format(self.client_id))
        self.needs_close = True

项目：emscripten-docker 作者：apiaryio | 项目源码 | 文件源码

def get_tags(versions):
    logging.info("Reading tags from repo %s" % (TAGS_URL))
    r = requests.get(TAGS_URL)
    if r.status_code != 200:
        raise RuntimeError(r.text)
    else:
        tags = [i['name'] for i in json.loads(
            r.text) if any(x in i['name'] for x in versions)]
        while 'next' in r.links:
            r = requests.get(r.links['next']['url'])
            if r.status_code != 200:
                raise RuntimeError(r.text)
            else:
                tags.extend([i['name'] for i in json.loads(r.text)
                             if any(x in i['name'] for x in versions)])
    return tags

项目：OldMunkiPackages 作者：aysiu | 项目源码 | 文件源码

def trash_old_stuff(trashlist, trashpath, newpath):
    if isinstance(trashlist, list):
        for old_location in trashlist:
            # Get the subfolders needed to be created
            path_within_destination=os.path.relpath(old_location, trashpath)
            # Create what will be the destination path
            new_location=os.path.join(newpath, path_within_destination)
            # Make sure all the relevant subfolders exist in the destination
            if not os.path.exists(os.path.dirname(new_location)):
                os.makedirs(os.path.dirname(new_location))
            # Even though we've been double-checking paths all along, let's just make one last check
            if os.path.exists(old_location) and os.path.isdir(newpath):
                os.rename(old_location, new_location)
                logging.info("Moving %s to %s\n" % (old_location, new_location))
            else:
                logging.error("One of %s or %s does not exist\n" % (old_location, new_location))
    else:
        logging.error("%s is not a valid list\n" % trashlist)

# Function that checks paths are writable

项目：PyPlanet 作者：PyPlanet | 项目源码 | 文件源码

def create_from_settings(cls, instance, conf):
        try:
            engine_path, _, cls_name = conf['ENGINE'].rpartition('.')
            db_name = conf['NAME']
            db_options = conf['OPTIONS'] if 'OPTIONS' in conf and conf['OPTIONS'] else dict()

            # FIX for #331. Replace utf8 by utf8mb4 in the mysql driver encoding.
            if conf['ENGINE'] == 'peewee_async.MySQLDatabase' and 'charset' in db_options and db_options['charset'] == 'utf8':
                logging.info('Forcing to use \'utf8mb4\' instead of \'utf8\' for the MySQL charset option! (Fix #331).')
                db_options['charset'] = 'utf8mb4'

            # We will try to load it so we have the validation inside this class.
            engine = getattr(importlib.import_module(engine_path), cls_name)
        except ImportError:
            raise ImproperlyConfigured('Database engine doesn\'t exist!')
        except Exception as e:
            raise ImproperlyConfigured('Database configuration isn\'t complete or engine could\'t be found!')

        return cls(engine, instance, db_name, **db_options)