`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`(url)
`get_session`(url)	Function that determines the correct Session object to be returned
`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)	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, caused_by=None, **kwargs)#

Bases: Exception

Common base class for all non-exit exceptions.

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

return_code = 1#

reportable = False#

__repr__()#: Return repr(self).

__str__()#: Return str(self).

dump_map()#

stringify(obj, content_max_len=0)#

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

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__()#: Return repr(self).

_make_single_record(package_filename)#

static _dedupe_pkgs_dir_contents(pkgs_dir_contents)#

exception CondaDependencyError(message)#

Bases: conda.CondaError

Common base class for all non-exit exceptions.

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

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

Bases: conda.CondaError

Common base class for all non-exit exceptions.

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

exception CondaSSLError(message, caused_by=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, errno, **kwargs)#

Bases: conda.CondaError, OSError

Common base class for all non-exit exceptions.

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

exception ProxyError#

Bases: conda.CondaError

Common base class for all non-exit exceptions.

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

exception UnavailableInvalidChannel(channel, status_code, 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#

class Channel(scheme=None, auth=None, location=None, token=None, name=None, platform=None, package_filename=None)#

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

Package Spec: 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#

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_url, name=None)#

urls(with_credentials=False, subdirs=None)#

url(with_credentials=False)#

__str__()#: Return str(self).

__repr__()#: Return repr(self).

__eq__(other)#: Return self==value.

__hash__()#: Return hash(self).

__nonzero__()#

__bool__()#

__json__()#

dump()#

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

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

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.

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

property cache_path_json#

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

load(*, state_only=False) → str#

load_state()#

Update self.state without reading repodata.json.

Return self.state.

save(data: str)#: Write data to <repodata>.json cache path, synchronize state.

replace(temp_path: pathlib.Path)#

Rename path onto <repodata>.json 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`#