`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`

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

Bases: Exception

Common base class for all non-exit exceptions.

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_

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.

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.

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

Bases: conda.CondaError

Common base class for all non-exit exceptions.

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

Bases: conda.CondaError, OSError

Common base class for all non-exit exceptions.

exception ProxyError

Bases: conda.CondaError

Common base class for all non-exit exceptions.

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

Bases: ChannelError

Common base class for all non-exit exceptions.

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

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'

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

exception RepodataOnDisk

Bases: Exception

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

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.

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: Path | str = '', cache_path_state: 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.

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

__missing__(key: str)

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.

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.

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

create_cache_dir()

repodata

Classes

Functions

Attributes

`repodata`