swh.lister.launchpad.lister module#

class swh.lister.launchpad.lister.LaunchpadListerState(git_date_last_modified: datetime | None = None, bzr_date_last_modified: datetime | None = None)[source]#

Bases: object

State of Launchpad lister

git_date_last_modified: datetime | None = None#: modification date of last updated git repository since last listing

bzr_date_last_modified: datetime | None = None#: modification date of last updated bzr repository since last listing

swh.lister.launchpad.lister.origin(vcs_type: str, repo: Any) → str[source]#: Determine the origin url out of a repository with a given vcs_type

swh.lister.launchpad.lister.retry_if_restful_error(retry_state)[source]#

class swh.lister.launchpad.lister.LaunchpadLister(scheduler: SchedulerInterface, url: str = 'https://launchpad.net/', instance: str = 'launchpad', incremental: bool = False, 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)[source]#

Bases: Lister[LaunchpadListerState, Tuple[str, Collection]]

List repositories from Launchpad (git or bzr).

Parameters:

scheduler – instance of SchedulerInterface
incremental – defines if incremental listing should be used, in that case only modified or new repositories since last incremental listing operation will be returned

LAUNCHPAD_URL = 'https://launchpad.net/'#

LISTER_NAME: str = 'launchpad'#

state_from_dict(d: Dict[str, Any]) → LaunchpadListerState[source]#: Convert the state stored in the scheduler backend (as a dict), to the concrete StateType for this lister.

state_to_dict(state: LaunchpadListerState) → Dict[str, Any][source]#

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.

get_pages() → Iterator[Tuple[str, Collection]][source]#: Yields an iterator on all git/bzr repositories hosted on Launchpad sorted by last modification date in ascending order.

get_origins_from_page(page: Tuple[str, Collection]) → Iterator[ListedOrigin][source]#: Iterate on all git repositories and yield ListedOrigin instances.

finalize() → None[source]#

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.