`repodata`#

Repodata interface.

Classes#

`PackageCacheData`
`Channel`	Channel:
`RepoInterface`	Helper class that provides a standard way to create an ABC using
`CondaRepoInterface`	Provides an interface for retrieving repodata data from channels.
`RepodataState`	Load/save info file that accompanies cached repodata.json.
`RepodataCache`	Handle caching for a single repodata.json + repodata.info.json
`RepodataFetch`	Combine RepodataCache and RepoInterface to provide subdir_data.SubdirData()

Functions#

`stringify`(obj[, content_max_len])
`maybe_unquote`(→ str)
`get_session`(url)	Function that determines the correct Session object to be returned
`mkdir_p_sudo_safe`(path)
`lock`(fd, *[, lock_attempts])
`get_repo_interface`(→ type[RepoInterface])
`_add_http_value_to_dict`(resp, http_key, d, dict_key)
`conda_http_errors`(url, repodata_fn)	Use in a with: statement to translate requests exceptions to conda ones.
`_md5_not_for_security`(data)
`cache_fn_url`(url[, repodata_fn])
`get_cache_control_max_age`(cache_control_value)
`create_cache_dir`()

Attributes#

`CONDA_HOMEPAGE_URL`
`REPODATA_FN`
`context`
`join_url`
`stderrlog`
`CHECK_ALTERNATE_FORMAT_INTERVAL`
`LAST_MODIFIED_KEY`
`ETAG_KEY`
`CACHE_CONTROL_KEY`
`URL_KEY`
`CACHE_STATE_SUFFIX`
`ERROR_SNIPPET_LENGTH`

exception CondaError(message: str | None, caused_by: Any = None, **kwargs)#

Bases: Exception

Common base class for all non-exit exceptions.

Initialize self. See help(type(self)) for accurate signature.

return_code: int = 1#

reportable: bool = False#

__repr__() → str#: Return repr(self).

__str__() → str#: Return str(self).

dump_map() → dict[str, Any]#

stringify(obj, content_max_len=0)#

CONDA_HOMEPAGE_URL: Final = 'https://conda.io'#

REPODATA_FN: Final = 'repodata.json'#

context#

join_url#

maybe_unquote(url: str) → str#

class PackageCacheData(pkgs_dir)#

property _package_cache_records#

property is_writable#

_cache_: dict[str, PackageCacheData]#

insert(package_cache_record)#

load()#

reload()#

get(package_ref: conda.models.records.PackageRecord, 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__()#: Return repr(self).

_make_single_record(package_filename)#

static _dedupe_pkgs_dir_contents(pkgs_dir_contents)#

exception CondaDependencyError(message: str)#

Bases: conda.CondaError

Common base class for all non-exit exceptions.

Initialize self. See help(type(self)) for accurate signature.

exception CondaHTTPError(message: str, url: str, status_code: int | str, reason: str, elapsed_time: datetime.timedelta | str, response: requests.Response | None = None, caused_by: Any = None)#

Bases: conda.CondaError

Common base class for all non-exit exceptions.

Initialize self. See help(type(self)) for accurate signature.

exception CondaSSLError(message: str | None, caused_by: Any | None = None, **kwargs)#

Bases: conda.CondaError

Common base class for all non-exit exceptions.

Initialize self. See help(type(self)) for accurate signature.

exception NotWritableError(path: conda.common.path.PathType, errno: int, **kwargs)#

Bases: conda.CondaError, OSError

Common base class for all non-exit exceptions.

Initialize self. See help(type(self)) for accurate signature.

exception ProxyError(message: str | None = None)#

Bases: conda.CondaError

Common base class for all non-exit exceptions.

Initialize self. See help(type(self)) for accurate signature.

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

Bases: ChannelError

Common base class for all non-exit exceptions.

Initialize self. See help(type(self)) for accurate signature.

status_code: str | int#

Channel: scheme <> auth <> location <> token <> channel <> subchannel <> platform <> package_filename

Package Spec: channel <> subchannel <> namespace <> package_name

property channel_location: str | None#

property channel_name: str#

property subdir: str | None#

property canonical_name: str#

property base_url: str | None#

property base_urls: tuple[str | None, Ellipsis]#

property subdir_url: str#

property channels: tuple[Channel, Ellipsis]#

property url_channel_wtf: tuple[str | None, str]#

_cache_#

static _reset_state() → None#

static from_url(url: str) → Channel#

static from_channel_name(channel_name: str) → Channel#

static from_value(value: str | None) → Channel#

Construct a new Channel from a single value.

Parameters:

value --

Anyone of the following forms:

None, or one of the special strings "<unknown>", "None:///<unknown>", or "None":: represents the unknown channel, used for packages with unknown origin.
A URL including a scheme like file:// or https://:: represents a channel URL.
A local directory path:: represents a local channel; relative paths must start with ./.
A package file (i.e. the path to a file ending in .conda or .tar.bz2):: represents a channel for a single package
A known channel name:: represents a known channel, e.g. from the users .condarc file or the global configuration.

Returns:

A channel object.

static make_simple_channel(channel_alias: Channel, channel_url: str, name: str | None = None) → Channel#

urls(with_credentials: bool = False, subdirs: collections.abc.Iterable[str] | None = None) → list[str]#

Generate URLs for this channel across specified platforms.

Parameters:

with_credentials -- If True, include authentication credentials (token, auth) in URLs.
subdirs -- Specific platform subdirs to generate URLs for. If None, uses the channel's platform (if defined) or falls back to context.subdirs. If this is explicitly provided, overrides any platform defined in the channel.

Examples

>>> channel = Channel("conda-forge")
>>> channel.urls()  # Uses context.subdirs
['https://conda.anaconda.org/conda-forge/linux-64',
'https://conda.anaconda.org/conda-forge/noarch']

>>> channel = Channel("conda-forge/linux-aarch64")
>>> channel.urls()  # Uses channel's platform
['https://conda.anaconda.org/conda-forge/linux-aarch64',
'https://conda.anaconda.org/conda-forge/noarch']

>>> channel.urls(subdirs=("osx-64", "noarch"))
['https://conda.anaconda.org/conda-forge/osx-64',
'https://conda.anaconda.org/conda-forge/noarch']

Returns:: List of URLs for accessing this channel's specified subdirectories.
Return type:: list[str]

url(with_credentials: bool = False) → str | None#

__str__() → str#: Return str(self).

__repr__() → str#: Return repr(self).

__eq__(other: Any) → bool#: Return self==value.

__hash__() → int#: Return hash(self).

__nonzero__() → bool#

__bool__() → bool#

__json__() → dict[str, Any]#

dump() → dict[str, Any]#

get_session(url: str)#: Function that determines the correct Session object to be returned based on the URL that is passed in.

mkdir_p_sudo_safe(path)#

lock(fd, *, lock_attempts=LOCK_ATTEMPTS)#

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: conda.models.channel.Channel | str, status_code: str | int, response: requests.models.Response | None = None)#

Bases: conda.exceptions.UnavailableInvalidChannel

Subclass used to determine when empty repodata should be cached, e.g. for a channel that doesn't provide current_repodata.json

Initialize self. See help(type(self)) for accurate signature.

exception RepodataOnDisk#

Bases: Exception

Indicate that RepoInterface.repodata() successfully wrote repodata to disk, instead of returning a string.

Initialize self. See help(type(self)) for accurate signature.

class RepoInterface#

Bases: abc.ABC

Helper class that provides a standard way to create an ABC using inheritance.

repodata(state: dict) → str#: Given a mutable state dictionary with information about the cache, return repodata.json (or current_repodata.json) as a str. This function also updates state, which is expected to be saved by the caller.

exception Response304ContentUnchanged#

Bases: Exception

Common base class for all non-exit exceptions.

Initialize self. See help(type(self)) for accurate signature.

get_repo_interface() → type[RepoInterface]#

class CondaRepoInterface(url: str, repodata_fn: str | None, **kwargs)#

Bases: RepoInterface

Provides an interface for retrieving repodata data from channels.

_url: str#

_repodata_fn: str#

repodata(state: RepodataState) → str | None#: Given a mutable state dictionary with information about the cache, return repodata.json (or current_repodata.json) as a str. This function also updates state, which is expected to be saved by the caller.

_add_http_value_to_dict(resp, http_key, d, dict_key)#

conda_http_errors(url, repodata_fn)#: Use in a with: statement to translate requests exceptions to conda ones.

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

Bases: collections.UserDict

Load/save info file that accompanies cached repodata.json.

property mod: str#: Last-Modified header or ""

property etag: str#: Etag header or ""

property cache_control: str#: Cache-Control header or ""

_aliased#

_strings#

has_format(format: str) → tuple[bool, datetime.datetime | None]#

set_has_format(format: str, value: bool)#

clear_has_format(format: str)#: Remove 'has_{format}' instead of setting to False.

should_check_format(format: str) → bool#: Return True if named format should be attempted.

__contains__(key: str) → bool#

__setitem__(key: str, item: Any) → None#

__getitem__(key: str) → Any#

class RepodataCache(base, repodata_fn)#

Handle caching for a single repodata.json + repodata.info.json (<hex-string>*.json inside dir)

Avoid race conditions while loading, saving repodata.json and cache state.

Also support bytes as in repodata_shards.msgpack.zst

base: directory and filename prefix for cache, e.g. /cache/dir/abc123; writes /cache/dir/abc123.json

property cache_path_json#

property cache_path_shards#

property cache_path_state#: Out-of-band etag and other state needed by the RepoInterface.

load(*, state_only=False, binary=False) → str | bytes#

load_state(binary=False)#

Update self.state without reading repodata.

Return self.state.

save(data: str | bytes)#: Write data to <repodata> cache path, by calling self.replace().

replace(temp_path: pathlib.Path, target=None)#

Rename path onto <repodata> path, synchronize state.

Relies on path's mtime not changing on move. temp_path should be adjacent to self.cache_path_json to be on the same filesystem.

refresh(refresh_ns=0)#: Update access time in cache info file to indicate a HTTP 304 Not Modified response.

lock(mode='a+')#

Lock .info.json file. Hold lock while modifying related files.

mode: "a+" then seek(0) to write/create; "r+" to read.

stale()#: Compare state refresh_ns against cache control header and context.local_repodata_ttl.

timeout()#: Return number of seconds until cache times out (<= 0 if already timed out).

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

Combine RepodataCache and RepoInterface to provide subdir_data.SubdirData() with what it needs.

Provide a variety of formats since some RepoInterface have to json.loads(...) anyway, and some clients don't need the Python data structure at all.

property url_w_repodata_fn#

property cache_path_json#

property cache_path_state#: Out-of-band etag and other state needed by the RepoInterface.

property repo_cache: RepodataCache#

property _repo: RepoInterface#: Changes as we mutate 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]#

Retrieve parsed latest or latest-cached repodata as a dict; update cache.

Returns:: (repodata contents, state including cache headers)

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

Retrieve latest or latest-cached repodata; update cache.

Returns:: (pathlib.Path to uncompressed repodata contents, RepodataState)

fetch_latest() → tuple[dict | str, RepodataState]#: Return up-to-date repodata and cache information. Fetch repodata from remote if cache has expired; return cached data if cache has not expired; return stale cached data or dummy data if in offline mode.

read_cache() → tuple[str, RepodataState]#: Read repodata from disk, without trying to fetch a fresh version.

_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()#

repodata#

Classes#

Functions#

Attributes#

`repodata`#