• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号
登陆 注册
描述文字

Python utils.read_json_from_uri函数代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Python中stackalytics.processor.utils.read_json_from_uri函数的典型用法代码示例。如果您正苦于以下问题：Python read_json_from_uri函数的具体用法？Python read_json_from_uri怎么用？Python read_json_from_uri使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。



在下文中一共展示了read_json_from_uri函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: lp_bug_generator

def lp_bug_generator(module, modified_since):
    uri = LP_URI_DEVEL % (module + '?ws.op=searchTasks')
    for status in BUG_STATUSES:
        uri += '&status=' + six.moves.urllib.parse.quote_plus(status)
    if modified_since:
        uri += '&modified_since=' + utils.timestamp_to_utc_date(modified_since)

    while uri:
        LOG.debug('Reading chunk from uri %s', uri)
        chunk = utils.read_json_from_uri(uri, session=launchpad_session)

        if not chunk:
            LOG.warning('No data was read from uri %s', uri)
            break

        for record in chunk['entries']:
            yield record

            related_tasks_uri = record['related_tasks_collection_link']
            LOG.debug('Reading related task from uri %s', related_tasks_uri)
            related_tasks = utils.read_json_from_uri(related_tasks_uri,
                                                     session=launchpad_session)
            if not related_tasks:
                LOG.warning('No data was read from uri %s', uri)
            elif related_tasks['entries']:
                for related_task in related_tasks['entries']:
                    yield related_task

        uri = chunk.get('next_collection_link')
开发者ID:openstack，项目名称:stackalytics，代码行数:29，代码来源:launchpad_utils.py


示例2: _zanata_get_project_versions

def _zanata_get_project_versions(project_id):
    LOG.debug("Reading iterations for project %s" % project_id)
    uri = ZANATA_URI % ('projects/p/%s' % project_id)
    project_data = utils.read_json_from_uri(uri, session=zanata_session)

    return (it['id'] for it in project_data.get('iterations', [])
            if ZANATA_VERSION_PATTERN.match(it['id']))
开发者ID:ekcs，项目名称:stackalytics，代码行数:7，代码来源:zanata.py


示例3: _zanata_get_projects

def _zanata_get_projects():
    uri = ZANATA_URI % 'projects'
    LOG.debug("Reading projects from %s" % uri)
    projects_data = utils.read_json_from_uri(uri, session=zanata_session)

    return (p['id'] for p in projects_data
            if ZANATA_PROJECT_PATTERN.match(p['id']))
开发者ID:ekcs，项目名称:stackalytics，代码行数:7，代码来源:zanata.py


示例4: _update_with_driverlog_data

def _update_with_driverlog_data(default_data, driverlog_data_uri):
    LOG.info('Reading DriverLog data from uri: %s', driverlog_data_uri)
    driverlog_data = utils.read_json_from_uri(driverlog_data_uri)

    if not driverlog_data:
        LOG.warning('DriverLog data is not available')
        return

    module_cis = collections.defaultdict(list)
    for driver in driverlog_data['drivers']:
        if 'ci' not in driver:
            continue

        module = (driver.get('repo') or driver['project_id']).split('/')[1]

        module_cis[module].append(driver)

        default_data['users'].append({
            'user_id': user_processor.make_user_id(ci_id=driver['name']),
            'user_name': driver['name'],
            'static': True,
            'companies': [
                {'company_name': driver['vendor'], 'end_date': None}],
        })

    for repo in default_data['repos']:
        if repo['module'] in module_cis:
            repo['drivers'] = module_cis[repo['module']]
开发者ID:Mirantis，项目名称:stackalytics，代码行数:28，代码来源:default_data_processor.py


示例5: _update_with_driverlog_data

def _update_with_driverlog_data(default_data, driverlog_data_uri):
    LOG.info('Reading DriverLog data from uri: %s', driverlog_data_uri)
    driverlog_data = utils.read_json_from_uri(driverlog_data_uri)

    module_ci_ids = {}
    ci_ids = set()
    for driver in driverlog_data['drivers']:
        if 'ci' in driver:
            module = driver['project_id'].split('/')[1]

            if module not in module_ci_ids:
                module_ci_ids[module] = {}
            ci_id = driver['ci']['id']
            module_ci_ids[module][ci_id] = driver

            if ci_id not in ci_ids:
                ci_ids.add(ci_id)
                default_data['users'].append({
                    'user_id': user_processor.make_user_id(gerrit_id=ci_id),
                    'gerrit_id': ci_id,
                    'user_name': ci_id,
                    'static': True,
                    'companies': [
                        {'company_name': driver['vendor'], 'end_date': None}],
                })

    for repo in default_data['repos']:
        if repo['module'] in module_ci_ids:
            repo['ci'] = module_ci_ids[repo['module']]
开发者ID:OpenStackRussia，项目名称:stackalytics，代码行数:29，代码来源:default_data_processor.py


示例6: main

def main():
    # init conf and logging
    conf = cfg.CONF
    conf.register_cli_opts(config.OPTS)
    conf.register_opts(config.OPTS)
    conf()

    logging.setup('stackalytics')
    LOG.info('Logging enabled')

    runtime_storage_inst = runtime_storage.get_runtime_storage(
        cfg.CONF.runtime_storage_uri)

    default_data = utils.read_json_from_uri(cfg.CONF.default_data_uri)
    if not default_data:
        LOG.critical('Unable to load default data')
        return not 0
    default_data_processor.process(runtime_storage_inst,
                                   default_data,
                                   cfg.CONF.sources_root,
                                   cfg.CONF.force_update)

    process_program_list(runtime_storage_inst, cfg.CONF.program_list_uri)

    update_pids(runtime_storage_inst)

    record_processor_inst = record_processor.RecordProcessor(
        runtime_storage_inst)

    update_records(runtime_storage_inst, record_processor_inst)

    apply_corrections(cfg.CONF.corrections_uri, runtime_storage_inst)

    # long operation should be the last
    update_members(runtime_storage_inst, record_processor_inst)
开发者ID:vnaboychenko，项目名称:stackalytics，代码行数:35，代码来源:main.py


示例7: main

def main():
    utils.init_config_and_logging(config.CONNECTION_OPTS + config.PROCESSOR_OPTS)

    runtime_storage_inst = runtime_storage.get_runtime_storage(cfg.CONF.runtime_storage_uri)

    default_data = utils.read_json_from_uri(cfg.CONF.default_data_uri)
    if not default_data:
        LOG.critical("Unable to load default data")
        return not 0

    default_data_processor.process(runtime_storage_inst, default_data, cfg.CONF.driverlog_data_uri)

    process_project_list(runtime_storage_inst, cfg.CONF.project_list_uri)

    update_pids(runtime_storage_inst)

    record_processor_inst = record_processor.RecordProcessor(runtime_storage_inst)

    process(runtime_storage_inst, record_processor_inst)

    apply_corrections(cfg.CONF.corrections_uri, runtime_storage_inst)

    # long operation should be the last
    update_members(runtime_storage_inst, record_processor_inst)

    runtime_storage_inst.set_by_key("runtime_storage_update_time", utils.date_to_timestamp("now"))
开发者ID:karthik-suresh，项目名称:stackalytics，代码行数:26，代码来源:main.py


示例8: main

def main():
    # init conf and logging
    conf = cfg.CONF
    conf.register_cli_opts(config.OPTS)
    conf.register_opts(config.OPTS)
    conf()

    logging.setup('stackalytics')
    LOG.info('Logging enabled')

    runtime_storage_inst = runtime_storage.get_runtime_storage(
        cfg.CONF.runtime_storage_uri)

    default_data = utils.read_json_from_uri(cfg.CONF.default_data_uri)
    if not default_data:
        LOG.critical('Unable to load default data')
        return not 0
    default_data_processor.process(runtime_storage_inst,
                                   default_data,
                                   cfg.CONF.sources_root)

    update_pids(runtime_storage_inst)

    update_repos(runtime_storage_inst)

    apply_corrections(cfg.CONF.corrections_uri, runtime_storage_inst)
开发者ID:dicortazar，项目名称:stackalytics，代码行数:26，代码来源:main.py


示例9: _openstack_profile_by_email

def _openstack_profile_by_email(email):
    LOG.debug('Lookup user email %s at OpenStackID', email)
    uri = OSID_URI % email
    data = utils.read_json_from_uri(uri, session=_openstackid_session)

    if not data:
        return None

    if not data.get('data'):
        return None  # not found

    return data['data'][-1]  # return the last (most recent) record
开发者ID:openstack，项目名称:stackalytics，代码行数:12，代码来源:openstackid_utils.py


示例10: lp_blueprint_generator

def lp_blueprint_generator(module):
    uri = LP_URI_DEVEL % (module + '/all_specifications')
    while uri:
        LOG.debug('Reading chunk from uri %s', uri)
        chunk = utils.read_json_from_uri(uri, session=launchpad_session)

        if not chunk:
            LOG.warning('No data was read from uri %s', uri)
            break

        for record in chunk['entries']:
            yield record

        uri = chunk.get('next_collection_link')
开发者ID:openstack，项目名称:stackalytics，代码行数:14，代码来源:launchpad_utils.py


示例11: apply_corrections

def apply_corrections(uri, runtime_storage_inst):
    LOG.info('Applying corrections from uri %s', uri)
    corrections = utils.read_json_from_uri(uri)
    if not corrections:
        LOG.error('Unable to read corrections from uri: %s', uri)
        return

    valid_corrections = []
    for c in corrections['corrections']:
        if 'primary_key' in c:
            valid_corrections.append(c)
        else:
            LOG.warn('Correction misses primary key: %s', c)
    runtime_storage_inst.apply_corrections(valid_corrections)
开发者ID:pombredanne，项目名称:stackalytics，代码行数:14，代码来源:main.py


示例12: _get_lp_info

    def _get_lp_info(self, email):
        lp_profile = None
        if not re.match(r'[\w\d_\.-][email protected]([\w\d_\.-]+\.)+[\w]+', email):
            LOG.debug('User email is not valid %s' % email)
        else:
            LOG.debug('Lookup user email %s at Launchpad' % email)
            uri = ('https://api.launchpad.net/1.0/people/?'
                   'ws.op=getByEmail&email=%s' % email)
            lp_profile = utils.read_json_from_uri(uri)

        if not lp_profile:
            LOG.debug('User with email %s not found', email)
            return None, None

        return lp_profile['name'], lp_profile['display_name']
开发者ID:dicortazar，项目名称:stackalytics，代码行数:15，代码来源:record_processor.py


示例13: main

def main():
    # init conf and logging
    conf = cfg.CONF
    conf.register_cli_opts(config.OPTS)
    conf.register_opts(config.OPTS)
    conf(project='stackalytics')

    logging.setup('stackalytics')
    LOG.info('Logging enabled')

    runtime_storage_inst = runtime_storage.get_runtime_storage(
        cfg.CONF.runtime_storage_uri)

    default_data = utils.read_json_from_uri(cfg.CONF.default_data_uri)
    if not default_data:
        LOG.critical('Unable to load default data')
        return not 0

    gerrit = rcs.get_rcs(None, cfg.CONF.review_uri)
    gerrit.setup(key_filename=cfg.CONF.ssh_key_filename,
                 username=cfg.CONF.ssh_username)

    default_data_processor.process(runtime_storage_inst,
                                   default_data,
                                   cfg.CONF.git_base_uri,
                                   gerrit,
                                   cfg.CONF.driverlog_data_uri)

    process_program_list(runtime_storage_inst, cfg.CONF.program_list_uri)

    update_pids(runtime_storage_inst)

    record_processor_inst = record_processor.RecordProcessor(
        runtime_storage_inst)

    process(runtime_storage_inst, record_processor_inst)

    apply_corrections(cfg.CONF.corrections_uri, runtime_storage_inst)

    # long operation should be the last
    update_members(runtime_storage_inst, record_processor_inst)

    runtime_storage_inst.set_by_key('runtime_storage_update_time',
                                    utils.date_to_timestamp('now'))
开发者ID:Acidburn0zzz，项目名称:stackalytics，代码行数:44，代码来源:main.py


示例14: main

def main():
    # init conf and logging
    conf = cfg.CONF
    conf.register_cli_opts(config.OPTS)
    conf.register_opts(config.OPTS)
    logging.register_options(conf)
    logging.set_defaults()
    conf(project='stackalytics')

    logging.setup(conf, 'stackalytics')
    LOG.info('Logging enabled')
    conf.log_opt_values(LOG, std_logging.DEBUG)

    runtime_storage_inst = runtime_storage.get_runtime_storage(
        cfg.CONF.runtime_storage_uri)

    default_data = utils.read_json_from_uri(cfg.CONF.default_data_uri)
    if not default_data:
        LOG.critical('Unable to load default data')
        return not 0

    default_data_processor.process(runtime_storage_inst,
                                   default_data,
                                   cfg.CONF.driverlog_data_uri)

    process_project_list(runtime_storage_inst, cfg.CONF.project_list_uri)

    update_pids(runtime_storage_inst)

    record_processor_inst = record_processor.RecordProcessor(
        runtime_storage_inst)

    process(runtime_storage_inst, record_processor_inst)

    apply_corrections(cfg.CONF.corrections_uri, runtime_storage_inst)

    # long operation should be the last
    update_members(runtime_storage_inst, record_processor_inst)

    runtime_storage_inst.set_by_key('runtime_storage_update_time',
                                    utils.date_to_timestamp('now'))
开发者ID:alfred-huangjian，项目名称:stackalytics，代码行数:41，代码来源:main.py


示例15: _update_with_driverlog_data

def _update_with_driverlog_data(default_data, driverlog_data_uri):
    LOG.info('Reading DriverLog data from uri: %s', driverlog_data_uri)
    driverlog_data = utils.read_json_from_uri(driverlog_data_uri)

    cis = {}
    for driver in driverlog_data['drivers']:
        if 'ci' in driver:
            module = driver['project_id'].split('/')[1]

            if module not in cis:
                cis[module] = {}
            cis[module][driver['ci']['id']] = driver

            default_data['users'].append({
                'launchpad_id': driver['ci']['id'],
                'user_name': driver['ci']['id'],
                'companies': [
                    {'company_name': driver['vendor'], 'end_date': None}],
            })

    for repo in default_data['repos']:
        if repo['module'] in cis:
            repo['ci'] = cis[repo['module']]
开发者ID:Tan0，项目名称:stackalytics，代码行数:23，代码来源:default_data_processor.py


示例16: _lp_profile_by_email

def _lp_profile_by_email(email):
    LOG.debug('Lookup user email %s at Launchpad', email)
    uri = LP_URI_V1 % ('people/?ws.op=getByEmail&email=' + email)
    lp_profile = utils.read_json_from_uri(uri, session=launchpad_session)
    utils.validate_lp_display_name(lp_profile)
    return lp_profile
开发者ID:openstack，项目名称:stackalytics，代码行数:6，代码来源:launchpad_utils.py


示例17: _lp_profile_by_launchpad_id

def _lp_profile_by_launchpad_id(launchpad_id):
    LOG.debug('Lookup user id %s at Launchpad', launchpad_id)
    uri = LP_URI_V1 % ('~' + launchpad_id)
    lp_profile = utils.read_json_from_uri(uri, session=launchpad_session)
    utils.validate_lp_display_name(lp_profile)
    return lp_profile
开发者ID:openstack，项目名称:stackalytics，代码行数:6，代码来源:launchpad_utils.py


示例18: lp_profile_by_launchpad_id

def lp_profile_by_launchpad_id(launchpad_id):
    LOG.debug('Lookup user id %s at Launchpad', launchpad_id)
    uri = LP_URI_V1 % ('~' + launchpad_id)
    return utils.read_json_from_uri(uri)
开发者ID:HolySparky，项目名称:stackalytics，代码行数:4，代码来源:launchpad_utils.py


示例19: _zanata_get_user_stats

def _zanata_get_user_stats(zanata_user_id, start_date, end_date):
    uri = ZANATA_URI % ('stats/user/%s/%s..%s' % (zanata_user_id,
                        start_date, end_date))
    return utils.read_json_from_uri(uri, session=zanata_session)
开发者ID:yeweiasia，项目名称:stackalytics，代码行数:4，代码来源:zanata.py


示例20: lp_profile_by_email

def lp_profile_by_email(email):
    LOG.debug('Lookup user email %s at Launchpad', email)
    uri = LP_URI_V1 % ('people/?ws.op=getByEmail&email=' + email)
    return utils.read_json_from_uri(uri)
开发者ID:HolySparky，项目名称:stackalytics，代码行数:4，代码来源:launchpad_utils.py



注：本文中的stackalytics.processor.utils.read_json_from_uri函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。

鲜花
握手
雷人
路过
鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
More+
上一篇：
Python utils.store_user函数代码示例发布时间：2022-05-27
下一篇：
Python utils.load_user函数代码示例发布时间：2022-05-27
热门推荐
More+
热门话题
More+
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服（服务时间 9:00～18:00）

在线QQ客服
地址：深圳市南山区西丽大学城创智工业园
电邮：jeky_zhao#qq.com
移动电话：139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap