swh.lister.gogs.lister module

class swh.lister.gogs.lister.GogsListerPage(repos: List[Dict[str, Any]] | None = None, next_link: str | None = None)

Bases: object

repos: List[Dict[str, Any]] | None = None

next_link: str | None = None

class swh.lister.gogs.lister.GogsListerState(last_seen_next_link: str | None = None, last_seen_repo_id: int | None = None)

Bases: object

last_seen_next_link: str | None = None

Last link header (could be already visited) during an incremental pass.

last_seen_repo_id: int | None = None

Last repo id seen during an incremental pass.

class swh.lister.gogs.lister.GogsLister(scheduler: SchedulerInterface, url: str | None = None, instance: str | None = None, api_token: str | None = None, page_size: int = 50, credentials: Dict[str, Dict[str, List[Dict[str, str]]]] | None = None, max_origins_per_page: int | None = None, max_pages: int | None = None, enable_origins: bool = True)

Bases: Lister[GogsListerState, GogsListerPage]

List origins from the Gogs

Gogs API documentation: gogs/docs-api

The API may be protected behind authentication so credentials/API tokens can be provided.

The lister supports pagination and provides next page URL through the ‘next’ value of the ‘Link’ header. The default value for page size (‘limit’) is 10 but the maximum allowed value is 50.

Api can usually be found at the location: https://<host>/api/v1/repos/search

LISTER_NAME: str = 'gogs'

VISIT_TYPE = 'git'

REPO_LIST_PATH = 'repos/search'

build_url(instance: str) → str

Build gogs url out of the instance.

state_from_dict(d: Dict[str, Any]) → GogsListerState

Convert the state stored in the scheduler backend (as a dict), to the concrete StateType for this lister.

state_to_dict(state: GogsListerState) → Dict[str, Any]

Convert the StateType for this lister to its serialization as dict for storage in the scheduler.

Values must be JSON-compatible as that’s what the backend database expects.

page_request(url: str, params: Dict[str, Any]) → Tuple[Dict[str, Any], Dict[str, Any]]

classmethod extract_repos(body: Dict[str, Any]) → List[Dict[str, Any]]

get_pages() → Iterator[GogsListerPage]

Retrieve a list of pages of listed results. This is the main loop of the lister.

Returns:

an iterator of raw pages fetched from the platform currently being listed.

get_origins_from_page(page: GogsListerPage) → Iterator[ListedOrigin]

Convert a page of Gogs repositories into a list of ListedOrigins

commit_page(page: GogsListerPage) → None

Custom hook called after the current page has been committed in the scheduler backend.

This method can be used to update the state after a page of origins has been successfully recorded in the scheduler backend. If the new state should be recorded at the point the lister completes, the updated attribute must be set.

finalize() → None

Custom hook to finalize the lister state before returning from the main loop.

This method must set updated if the lister has done some work.

If relevant, this method can use :meth`get_state_from_scheduler` to merge the current lister state with the one from the scheduler backend, reducing the risk of race conditions if we’re running concurrent listings.

This method is called in a finally block, which means it will also run when the lister fails.