nvchecker.api — The source plugin API

exception nvchecker.api.TemporaryError(code, message, response)[source]

A temporary error (e.g. network error) happens.

exception nvchecker.api.HTTPError(code, message, response)[source]

An HTTP 4xx error happens

class nvchecker.api.BaseWorker(task_sem: Semaphore, result_q: Queue[RawResult], tasks: List[Tuple[str, Dict[str, Any]]], keymanager: KeyManager)[source]

The base class for defining Worker classes for source plugins.

task_sem: asyncio.Semaphore

This is the rate-limiting semaphore. Workers should acquire it while doing one unit of work.

result_q: Queue[RawResult]

Results should be put into this queue.

tasks: List[Tuple[str, Entry]]

A list of tasks for the Worker to complete. Every task consists of a tuple for the task name (table name in the configuration file) and the content of that table (as a dict).

keymanager: KeyManager

The KeyManager for retrieving keys from the keyfile.

abstract async run() None[source]

Run the tasks. Subclasses should implement this method.

class nvchecker.api.RawResult(name: str, version: VersionResult, conf: Entry)[source]

The unprocessed result from a check.

Create new instance of RawResult(name, version, conf)

name: str

The name (table name) of the entry.

version: None | str | RichResult | List[str | RichResult] | Exception

The result from the check.

conf: Dict[str, Any]

The entry configuration (table content) of the entry.

class nvchecker.api.RichResult(*, version: str, gitref: str | None = None, revision: str | None = None, url: str | None = None)[source]

RichResult(*, version: ‘str’, gitref: ‘Optional[str]’ = None, revision: ‘Optional[str]’ = None, url: ‘Optional[str]’ = None)

version: str
gitref: str | None = None
revision: str | None = None
url: str | None = None
class nvchecker.api.AsyncCache[source]

A cache for use with async functions.

cache: Dict[Hashable, Any]
lock: Lock
async get_json(url: str, *, headers: Dict[str, str] = {}) Any[source]

Get specified url and return the response content as JSON.

The returned data will be cached for reuse.

async get(key: Hashable, func: Callable[[Hashable], Coroutine[Any, Any, Any]]) Any[source]

Run async func and cache its return value by key.

The key should be hashable, and the function will be called with it as its sole argument. For multiple simultaneous calls with the same key, only one will actually be called, and others will wait and return the same (cached) value.

class nvchecker.api.KeyManager(file: Path | None)[source]

Manages data in the keyfile.

get_key(name: str, legacy_name: str | None = None) str | None[source]

Get the named key (token) in the keyfile.

exception nvchecker.api.GetVersionError(msg: LiteralString, **kwargs: Any)[source]

An error occurred while getting version information.

Raise this when a known bad situation happens.

  • msg – The error message.

  • kwargs – Arbitrary additional context for the error.

class nvchecker.api.EntryWaiter[source]
async wait(name: str) str[source]

Wait on the name entry and return its result (the version string)

set_result(name: str, value: str) None[source]
set_exception(name: str, e: Exception) None[source]
nvchecker.api.session: nvchecker.httpclient.base.BaseSession

The object to send out HTTP requests, respecting various options in the configuration entry.

class nvchecker.httpclient.base.Response(headers: Mapping[str, str], body: bytes)[source]

The response of an HTTP request.

body: bytes
headers: Mapping[str, str]

Convert response content to JSON.

class nvchecker.httpclient.base.BaseSession[source]

The base class for different HTTP backend.

setup(concurreny: int = 20, timeout: int = 20) None[source]
async head(*args, **kwargs)[source]

Shortcut for HEAD request.

async get(*args, **kwargs)[source]

Shortcut for GET request.

async post(*args, **kwargs)[source]

Shortcut for POST request.

async request(url: str, *, method: str, headers: Dict[str, str] = {}, follow_redirects: bool = True, params=(), json=None, body=None) Response[source]
nvchecker.api.proxy = <ContextVar name='proxy' default=None>
nvchecker.api.user_agent = <ContextVar name='user_agent' default='lilydjwg/nvchecker 2.15dev'>
nvchecker.api.tries = <ContextVar name='tries' default=1>
nvchecker.api.verify_cert = <ContextVar name='verify_cert' default=True>
nvchecker.api.entry_waiter: contextvars.ContextVar

This ContextVar contains an EntryWaiter instance for waiting on other entries.