我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用pkg_resources.working_set()。
def version_info(ctx=None): """Return version information just like --version does.""" from . import __version__ prog = ctx.find_root().info_name if ctx else APP_NAME version = __version__ try: import pkg_resources except ImportError: pass else: for dist in iter(pkg_resources.working_set): scripts = dist.get_entry_map().get('console_scripts') or {} for _, entry_point in iteritems(scripts): if entry_point.module_name == (__package__ + '.__main__'): version = dist.version break return VERSION_INFO % dict(prog=prog, version=version)
def check_package_exists(package, lib_dir): """Check if a package is installed globally or in lib_dir. Returns True when the requirement is met. Returns False when the package is not installed or doesn't meet req. """ try: req = pkg_resources.Requirement.parse(package) except ValueError: # This is a zip file req = pkg_resources.Requirement.parse(urlparse(package).fragment) # Check packages from lib dir if lib_dir is not None: if any(dist in req for dist in pkg_resources.find_distributions(lib_dir)): return True # Check packages from global + virtual environment # pylint: disable=not-an-iterable return any(dist in req for dist in pkg_resources.working_set)
def status_packages(request): config = [] for k in sorted(dir(settings)): if k == 'KEY': continue if k.startswith('_'): continue if k.upper() != k: continue config.append((k, getattr(settings, k))) return render_to_response('sentry/admin/status/packages.html', { 'modules': sorted([(p.project_name, p.version) for p in pkg_resources.working_set]), 'extensions': [ (p.get_title(), '%s.%s' % (p.__module__, p.__class__.__name__)) for p in plugins.all(version=None) ], }, request)
def use_virtual_env(self): """ Use the virtual environment """ if not self.init_env(): raise Exception("Unable to init virtual environment") activate_file = os.path.join(self.env_path, "bin/activate_this.py") if os.path.exists(activate_file): with open(activate_file) as f: code = compile(f.read(), activate_file, 'exec') exec(code, {"__file__": activate_file}) else: raise Exception("Unable to activate virtual environment because %s does not exist." % activate_file) # patch up pkg pkg_resources.working_set = pkg_resources.WorkingSet._build_master()
def cli(): """ Print the version and exit. """ distributions = [dist for dist in pkg_resources.working_set if dist.project_name.startswith('farmer')] for distribution in sorted(distributions): click.echo('{} {}'.format(distribution.project_name, distribution.version))
def freeze(args): for dist in pkg_resources.working_set: info = _get_info(dist.project_name) output = "{name}=={version}".format(**info) if info['sha']: output += " # git sha {sha}".format(**info) print(output)
def _directory_import(self): """ Import astropy_helpers from the given path, which will be added to sys.path. Must return True if the import succeeded, and False otherwise. """ # Return True on success, False on failure but download is allowed, and # otherwise raise SystemExit path = os.path.abspath(self.path) # Use an empty WorkingSet rather than the man # pkg_resources.working_set, since on older versions of setuptools this # will invoke a VersionConflict when trying to install an upgrade ws = pkg_resources.WorkingSet([]) ws.add_entry(path) dist = ws.by_key.get(DIST_NAME) if dist is None: # We didn't find an egg-info/dist-info in the given path, but if a # setup.py exists we can generate it setup_py = os.path.join(path, 'setup.py') if os.path.isfile(setup_py): with _silence(): run_setup(os.path.join(path, 'setup.py'), ['egg_info']) for dist in pkg_resources.find_distributions(path, True): # There should be only one... return dist return dist
def freeze(args): sorted_dists = sorted(pkg_resources.working_set, key=lambda dist: dist.project_name.lower()) for dist in sorted_dists: info = _get_info(dist.project_name) output = "{name}=={version}".format(**info) if info['sha']: output += " # git sha {sha}".format(**info) print(output)
def __init__(self, registry_name, *a, **kw): """ Arguments: registry_name The name of this registry. """ # The container for the resolved item. self.records = OrderedDict() self.registry_name = registry_name _working_set = kw.pop('_working_set', working_set) self.raw_entry_points = [] if _working_set is None else list( _working_set.iter_entry_points(self.registry_name)) self._init(*a, **kw)
def __init__( self, logger='calmjs', action_key=DEST_RUNTIME, working_set=default_working_set, package_name=None, description=None, *a, **kw): """ Keyword Arguments: logger The logger to enable for pretty logging. Default: the calmjs root logger action_key The destination key where the command will be stored. Under this key the target driver runtime will be stored, and it will be popped off first before passing rest of kwargs to it. working_set The working_set to use for this instance. Default: pkg_resources.working_set package_name The package name that this instance of runtime is for. Used for the version flag. description The description for this runtime. """ self.logger = logger self.action_key = action_key self.working_set = working_set self.description = description or self.__doc__ self.package_name = package_name super(BaseRuntime, self).__init__(*a, **kw)
def capture_env(self): return collections.OrderedDict( (d.project_name, d.version) for d in pkg_resources.working_set)
def print_results(hits, name_column_width=25, terminal_width=None): installed_packages = [p.project_name for p in pkg_resources.working_set] for hit in hits: name = hit['name'] summary = hit['summary'] or '' if terminal_width is not None: # wrap and indent summary to fit terminal summary = textwrap.wrap(summary, terminal_width - name_column_width - 5) summary = ('\n' + ' ' * (name_column_width + 3)).join(summary) line = '%s - %s' % (name.ljust(name_column_width), summary) try: logger.notify(line) if name in installed_packages: dist = pkg_resources.get_distribution(name) logger.indent += 2 try: latest = highest_version(hit['versions']) if dist.version == latest: logger.notify('INSTALLED: %s (latest)' % dist.version) else: logger.notify('INSTALLED: %s' % dist.version) logger.notify('LATEST: %s' % latest) finally: logger.indent -= 2 except UnicodeEncodeError: pass
def search_packages_info(query): """ Gather details from installed distributions. Print distribution name, version, location, and installed files. Installed files requires a pip generated 'installed-files.txt' in the distributions '.egg-info' directory. """ installed_packages = dict( [(p.project_name.lower(), p) for p in pkg_resources.working_set]) for name in query: normalized_name = name.lower() if normalized_name in installed_packages: dist = installed_packages[normalized_name] package = { 'name': dist.project_name, 'version': dist.version, 'location': dist.location, 'requires': [dep.project_name for dep in dist.requires()], } filelist = os.path.join( dist.location, dist.egg_name() + '.egg-info', 'installed-files.txt') if os.path.isfile(filelist): package['files'] = filelist yield package
def get_installed_distributions(local_only=True, skip=('setuptools', 'pip', 'python'), include_editables=True, editables_only=False): """ Return a list of installed Distribution objects. If ``local_only`` is True (default), only return installations local to the current virtualenv, if in a virtualenv. ``skip`` argument is an iterable of lower-case project names to ignore; defaults to ('setuptools', 'pip', 'python'). [FIXME also skip virtualenv?] If ``editables`` is False, don't report editables. If ``editables_only`` is True , only report editables. """ if local_only: local_test = dist_is_local else: local_test = lambda d: True if include_editables: editable_test = lambda d: True else: editable_test = lambda d: not dist_is_editable(d) if editables_only: editables_only_test = lambda d: dist_is_editable(d) else: editables_only_test = lambda d: True return [d for d in pkg_resources.working_set if local_test(d) and d.key not in skip and editable_test(d) and editables_only_test(d) ]
def get_installed_distributions(local_only=True, skip=('setuptools', 'pip', 'python', 'distribute'), include_editables=True, editables_only=False): """ Return a list of installed Distribution objects. If ``local_only`` is True (default), only return installations local to the current virtualenv, if in a virtualenv. ``skip`` argument is an iterable of lower-case project names to ignore; defaults to ('setuptools', 'pip', 'python'). [FIXME also skip virtualenv?] If ``editables`` is False, don't report editables. If ``editables_only`` is True , only report editables. """ if local_only: local_test = dist_is_local else: local_test = lambda d: True if include_editables: editable_test = lambda d: True else: editable_test = lambda d: not dist_is_editable(d) if editables_only: editables_only_test = lambda d: dist_is_editable(d) else: editables_only_test = lambda d: True return [d for d in pkg_resources.working_set if local_test(d) and d.key not in skip and editable_test(d) and editables_only_test(d) ]
def version(self): try: from pip._vendor import pkg_resources except ImportError: import pkg_resources return next((p.version for p in pkg_resources.working_set if p.project_name.lower() == 'instabot'), "No match")
def easter_fixture(): class EasterFixture: group_name = 'abc' stub_egg = EasterEgg() saved_working_set = pkg_resources.working_set pkg_resources.working_set = pkg_resources.WorkingSet() pkg_resources.working_set.add(EasterFixture.stub_egg) yield EasterFixture pkg_resources.working_set = saved_working_set
def add_to_working_set(self): """Adds this EasterEgg to the global pkg_resources.working_set object.""" pkg_resources.working_set.add(self, replace=True) return self
def collect_installed_distributions(): """Yield the normalized spec and the names of top_level modules of all installed packages.""" for distribution in pkg_resources.working_set: distribution_spec = str(distribution.as_requirement()) distribution_top_level = guess_top_level(distribution) yield distribution_spec, distribution_top_level
def _get_pkg_resource(filename, package, prefix): """Query pkg_resources for the location of the filename.""" requirement = pkg_resources.Requirement.parse(package) target = os.path.join(prefix, filename) try: return pkg_resources.resource_filename(requirement, target) except pkg_resources.DistributionNotFound: # It may be that the working set is not in sync (e.g. if sys.path was # manipulated). Try to reload it just in case. pkg_resources.working_set = pkg_resources.WorkingSet() try: return pkg_resources.resource_filename(requirement, target) except pkg_resources.DistributionNotFound: return None
def _install(self, requirements_list: []) -> None: """ Install requirements in the given requirements file """ requirements_file = self._gen_requirements_file(requirements_list) try: fdnum, path = tempfile.mkstemp() fd = os.fdopen(fdnum, "w+") fd.write(requirements_file) fd.close() cmd = [self.virtual_pip, "install", "-r", path] output = b"" try: output = subprocess.check_output(cmd, stderr=subprocess.STDOUT) except Exception: LOGGER.debug("%s: %s", cmd, output.decode()) LOGGER.debug("requirements: %s", requirements_file) raise else: LOGGER.debug("%s: %s", cmd, output.decode()) finally: if os.path.exists(path): os.remove(path) pkg_resources.working_set = pkg_resources.WorkingSet._build_master()
def load_modules(self): """ Load all existing modules """ mod_dir = os.path.join(self.__code_dir, MODULE_DIR) if os.path.exists(os.path.join(self.__code_dir, VERSION_FILE)): fd = open(os.path.join(self.__code_dir, VERSION_FILE), "r") self.__current_version = int(fd.read()) fd.close() pkg_resources.working_set = pkg_resources.WorkingSet._build_master() for py in glob.glob(os.path.join(mod_dir, "*.py")): if mod_dir in py: mod_name = py[len(mod_dir) + 1:-3] else: mod_name = py[:-3] source_code = "" with open(py, "r") as fd: source_code = fd.read().encode("utf-8") sha1sum = hashlib.new("sha1") sha1sum.update(source_code) hv = sha1sum.hexdigest() self._load_module(mod_name, py, hv)
def deploy_version(self, key, mod, persist=False): """ Deploy a new version of the modules :param version The version of the deployed modules :modules modules A list of module names and the hashes of the code files """ LOGGER.info("Deploying code (key=%s)" % key) # deploy the new code name = mod[1] source_code = mod[2] # if the module is new, or update if name not in self.__modules or key != self.__modules[name][0]: # write the new source source_file = os.path.join(self.__code_dir, MODULE_DIR, name + ".py") fd = open(source_file, "w+") fd.write(source_code) fd.close() # (re)load the new source self._load_module(name, source_file, key) if persist: with open(os.path.join(self.__code_dir, PERSIST_FILE), "w+") as fd: json.dump(mod, fd) pkg_resources.working_set = pkg_resources.WorkingSet._build_master()
