repodata#

Repodata 接口。

#

PackageCacheData

Channel

Channel

RepoInterface

提供使用 ABC 创建标准方法的帮助类

CondaRepoInterface

提供从频道检索 repodata 数据的接口。

RepodataState

加载/保存伴随缓存 repodata.json 的信息文件。

RepodataCache

处理单个 repodata.json + repodata.info.json 的缓存

RepodataFetch

组合 RepodataCache 和 RepoInterface 以提供 subdir_data.SubdirData()

函数#

stringify(obj[, content_max_len])

maybe_unquote(url)

get_session(url)

确定要返回的正确 Session 对象的函数

mkdir_p_sudo_safe(path)

lock(fd)

get_repo_interface(→ type[RepoInterface])

_add_http_value_to_dict(resp, http_key, d, dict_key)

conda_http_errors(url, repodata_fn)

在 with: 语句中使用,将请求异常转换为 conda 异常。

_md5_not_for_security(data)

cache_fn_url(url[, repodata_fn])

get_cache_control_max_age(cache_control_value)

create_cache_dir()

属性#

exception CondaError(message, caused_by=None, **kwargs)#

Bases: Exception

所有非退出异常的通用基类。

初始化 self。有关准确的签名,请参阅 help(type(self))。

return_code = 1#
reportable = False#
__repr__()#

返回 repr(self)。

__str__()#

返回 str(self)。

dump_map()#
stringify(obj, content_max_len=0)#
CONDA_HOMEPAGE_URL = 'https://docs.conda.org.cn'#
REPODATA_FN = 'repodata.json'#
context#
join_url#
maybe_unquote(url)#
class PackageCacheData(pkgs_dir)#
property _package_cache_records#
property is_writable#
_cache_: dict[str, PackageCacheData]#
insert(package_cache_record)#
load()#
reload()#
get(package_ref, default=NULL)#
remove(package_ref, default=NULL)#
query(package_ref_or_match_spec)#
iter_records()#
classmethod query_all(package_ref_or_match_spec, pkgs_dirs=None)#
classmethod first_writable(pkgs_dirs=None)#
classmethod writable_caches(pkgs_dirs=None)#
classmethod read_only_caches(pkgs_dirs=None)#
classmethod all_caches_writable_first(pkgs_dirs=None)#
classmethod get_all_extracted_entries()#
classmethod get_entry_to_link(package_ref)#
classmethod tarball_file_in_cache(tarball_path, md5sum=None, exclude_caches=())#
classmethod clear()#
tarball_file_in_this_cache(tarball_path, md5sum=None)#
_check_writable()#
static _clean_tarball_path_and_get_md5sum(tarball_path, md5sum=None)#
_scan_for_dist_no_channel(dist_str)#
itervalues()#
values()#
__repr__()#

返回 repr(self)。

_make_single_record(package_filename)#
static _dedupe_pkgs_dir_contents(pkgs_dir_contents)#
exception CondaDependencyError(message)#

基类: conda.CondaError

所有非退出异常的通用基类。

初始化 self。有关准确的签名,请参阅 help(type(self))。

exception CondaHTTPError(message, url, status_code, reason, elapsed_time, response=None, caused_by=None)#

基类: conda.CondaError

所有非退出异常的通用基类。

初始化 self。有关准确的签名,请参阅 help(type(self))。

exception CondaSSLError(message, caused_by=None, **kwargs)#

基类: conda.CondaError

所有非退出异常的通用基类。

初始化 self。有关准确的签名,请参阅 help(type(self))。

exception NotWritableError(path, errno, **kwargs)#

基类: conda.CondaError, OSError

所有非退出异常的通用基类。

初始化 self。有关准确的签名,请参阅 help(type(self))。

exception ProxyError#

基类: conda.CondaError

所有非退出异常的通用基类。

初始化 self。有关准确的签名,请参阅 help(type(self))。

exception UnavailableInvalidChannel(channel, status_code, response: requests.models.Response | None = None)#

基类: ChannelError

所有非退出异常的通用基类。

初始化 self。有关准确的签名,请参阅 help(type(self))。

status_code: str | int#
class Channel(scheme=None, auth=None, location=None, token=None, name=None, platform=None, package_filename=None)#

通道: scheme <> auth <> location <> token <> channel <> subchannel <> platform <> package_filename

包规范: channel <> subchannel <> namespace <> package_name

property channel_location#
property channel_name#
property subdir#
property canonical_name#
property base_url#
property base_urls#
property subdir_url#
property url_channel_wtf#
_cache_#
static _reset_state()#
static from_url(url)#
static from_channel_name(channel_name)#
static from_value(value: str | None) Self#

从单个值构造一个新的 Channel

参数:

value --

以下形式之一

None,或特殊字符串 "<unknown>"、"None:///<unknown>" 或 "None" 之一

代表未知通道,用于来源不明的软件包。

包含方案的 URL,例如 file://https://

代表通道 URL。

本地目录路径

代表本地通道;相对路径必须以 ./ 开头。

软件包文件(即以 .conda.tar.bz2 结尾的文件路径)

代表单个软件包的通道

已知的通道名称

代表已知的通道,例如来自用户的 .condarc 文件或全局配置。

返回:

一个通道对象。

static make_simple_channel(channel_alias, channel_url, name=None)#
urls(with_credentials=False, subdirs=None)#
url(with_credentials=False)#
__str__()#

返回 str(self)。

__repr__()#

返回 repr(self)。

__eq__(other)#

返回 self==value。

__hash__()#

返回 hash(self)。

__nonzero__()#
__bool__()#
__json__()#
dump()#
get_session(url: str)#

此函数基于传入的 URL 确定要返回的正确 Session 对象。

mkdir_p_sudo_safe(path)#
lock(fd)#
stderrlog#
CHECK_ALTERNATE_FORMAT_INTERVAL#
LAST_MODIFIED_KEY = 'mod'#
ETAG_KEY = 'etag'#
CACHE_CONTROL_KEY = 'cache_control'#
URL_KEY = 'url'#
CACHE_STATE_SUFFIX = '.info.json'#
ERROR_SNIPPET_LENGTH = 32#
exception RepodataIsEmpty(channel, status_code, response: requests.models.Response | None = None)#

基类: conda.exceptions.UnavailableInvalidChannel

用于确定何时应缓存空 repodata 的子类,例如对于不提供 current_repodata.json 的通道

初始化 self。有关准确的签名,请参阅 help(type(self))。

exception RepodataOnDisk#

Bases: Exception

指示 RepoInterface.repodata() 成功将 repodata 写入磁盘,而不是返回字符串。

初始化 self。有关准确的签名,请参阅 help(type(self))。

class RepoInterface#

基类: abc.ABC

帮助程序类,提供使用继承创建 ABC 的标准方法。

repodata(state: dict) str#

给定一个包含缓存信息的 mutable 状态字典,返回 repodata.json(或 current_repodata.json)作为 str。此函数还会更新状态,状态应由调用方保存。

exception Response304ContentUnchanged#

Bases: Exception

所有非退出异常的通用基类。

初始化 self。有关准确的签名,请参阅 help(type(self))。

get_repo_interface() type[RepoInterface]#
class CondaRepoInterface(url: str, repodata_fn: str | None, **kwargs)#

基类: RepoInterface

提供从频道检索 repodata 数据的接口。

_url: str#
_repodata_fn: str#
repodata(state: RepodataState) str | None#

给定一个包含缓存信息的 mutable 状态字典,返回 repodata.json(或 current_repodata.json)作为 str。此函数还会更新状态,状态应由调用方保存。

_add_http_value_to_dict(resp, http_key, d, dict_key)#
conda_http_errors(url, repodata_fn)#

在 with: 语句中使用,将请求异常转换为 conda 异常。

class RepodataState(cache_path_json: pathlib.Path | str = '', cache_path_state: pathlib.Path | str = '', repodata_fn='', dict=None)#

基类: collections.UserDict

加载/保存伴随缓存 repodata.json 的信息文件。

property mod: str#

Last-Modified 头部或 ""

property etag: str#

Etag 头部或 ""

property cache_control: str#

Cache-Control 头部或 ""

_aliased#
_strings#
has_format(format: str) tuple[bool, datetime.datetime | None]#
set_has_format(format: str, value: bool)#
clear_has_format(format: str)#

移除 'has_{format}' 而不是设置为 False。

should_check_format(format: str) bool#

如果指定的格式应该尝试,则返回 True。

__contains__(key: str) bool#
__setitem__(key: str, item: Any) None#
__getitem__(key: str) Any#
class RepodataCache(base, repodata_fn)#

处理单个 repodata.json + repodata.info.json 的缓存(dir 内的 <hex-string>*.json)

避免在加载、保存 repodata.json 和缓存状态时出现竞争条件。

base: 缓存的目录和文件名前缀,例如 /cache/dir/abc123;写入 /cache/dir/abc123.json

property cache_path_json#
property cache_path_state#

带外 etag 和 RepoInterface 所需的其他状态。

load(*, state_only=False) str#
load_state()#

更新 self.state 而不读取 repodata.json。

返回 self.state。

save(data: str)#

将数据写入 <repodata>.json 缓存路径,同步状态。

replace(temp_path: pathlib.Path)#

将路径重命名到 <repodata>.json 路径,同步状态。

依赖于路径的 mtime 在移动时不会更改。temp_path 应该与 self.cache_path_json 相邻,以便在同一文件系统上。

refresh(refresh_ns=0)#

更新缓存信息文件中的访问时间,以指示 HTTP 304 Not Modified 响应。

lock(mode='a+')#

锁定 .info.json 文件。在修改相关文件时保持锁定。

mode: "a+" 然后 seek(0) 进行写入/创建;"r+" 进行读取。

stale()#

将状态 refresh_ns 与缓存控制头部和 context.local_repodata_ttl 进行比较。

timeout()#

返回缓存超时的秒数(如果已超时,则 <= 0)。

class RepodataFetch(cache_path_base: pathlib.Path, channel: conda.models.channel.Channel, repodata_fn: str, *, repo_interface_cls)#

组合 RepodataCache 和 RepoInterface,为 subdir_data.SubdirData() 提供所需内容。

提供多种格式,因为某些 RepoInterface 必须无论如何都进行 json.loads(...),并且某些客户端根本不需要 Python 数据结构。

property url_w_repodata_fn#
property cache_path_json#
property cache_path_state#

带外 etag 和 RepoInterface 所需的其他状态。

property repo_cache: RepodataCache#
property _repo: RepoInterface#

当我们改变 self.repodata_fn 时会发生变化。

cache_path_base: pathlib.Path#
channel: conda.models.channel.Channel#
repodata_fn: str#
url_w_subdir: str#
url_w_credentials: str#
repo_interface_cls: Any#
fetch_latest_parsed() tuple[dict, RepodataState]#

检索已解析的最新或最新缓存的 repodata 作为字典;更新缓存。

返回:

(repodata 内容,状态包括缓存标头)

fetch_latest_path() tuple[pathlib.Path, RepodataState]#

检索最新或最新缓存的 repodata;更新缓存。

返回:

(pathlib.Path 指向未压缩的 repodata 内容,RepodataState)

fetch_latest() tuple[dict | str, RepodataState]#

返回最新的 repodata 和缓存信息。如果缓存已过期,则从远程获取 repodata;如果缓存未过期,则返回缓存数据;如果在离线模式下,则返回过时的缓存数据或虚拟数据。

read_cache() tuple[str, RepodataState]#

从磁盘读取 repodata,而不尝试获取新版本。

_md5_not_for_security(data)#
cache_fn_url(url, repodata_fn=REPODATA_FN)#
get_cache_control_max_age(cache_control_value: str | None)#
create_cache_dir()#