nvchecker.api — The source plugin API

exception nvchecker.api.TemporaryError(code, message, response)

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

exception nvchecker.api.HTTPError(code, message, response)

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)

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

Run the tasks. Subclasses should implement this method.

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

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)

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

A cache for use with async functions.

cache: Dict[Hashable, Any]

lock: Lock

async get_json(url: str, *, headers: Dict[str, str] = {}) → Any

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

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)

Manages data in the keyfile.

get_key(name: str, legacy_name: str | None = None) → str | None

Get the named key (token) in the keyfile.

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

An error occurred while getting version information.

Raise this when a known bad situation happens.

Parameters:

• msg – The error message.

• kwargs – Arbitrary additional context for the error.

class nvchecker.api.EntryWaiter

async wait(name: str) → str

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

set_result(name: str, value: str) → None

set_exception(name: str, e: Exception) → None

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)

The response of an HTTP request.

body: bytes

headers: Mapping[str, str]

json()

Convert response content to JSON.

class nvchecker.httpclient.base.BaseSession

The base class for different HTTP backend.

setup(concurreny: int = 20, timeout: int = 20) → None

async head(*args, **kwargs)

Shortcut for HEAD request.

async get(*args, **kwargs)

Shortcut for GET request.

async post(*args, **kwargs)

Shortcut for POST request.

async request(url: str, *, method: str, headers: Dict[str, str] = {}, follow_redirects: bool = True, params=(), json=None, body=None) → Response

nvchecker.api.proxy = <ContextVar name='proxy' default=None>

nvchecker.api.user_agent = <ContextVar name='user_agent' default='lilydjwg/nvchecker 2.16dev'>

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.