Changeset View
Changeset View
Standalone View
Standalone View
swh/lister/opam/lister.py
| Show First 20 Lines • Show All 45 Lines • ▼ Show 20 Lines | def __init__( | ||||
| url: str, | url: str, | ||||
| instance: str, | instance: str, | ||||
| credentials: CredentialsType = None, | credentials: CredentialsType = None, | ||||
| ): | ): | ||||
| super().__init__( | super().__init__( | ||||
| scheduler=scheduler, credentials=credentials, url=url, instance=instance, | scheduler=scheduler, credentials=credentials, url=url, instance=instance, | ||||
| ) | ) | ||||
| self.env = os.environ.copy() | self.env = os.environ.copy() | ||||
| self.env["OPAMROOT"] = tempfile.mkdtemp(prefix="swh_opam_lister") | self.opamroot = tempfile.mkdtemp(prefix="swh_opam_lister") | ||||
| call( | call( | ||||
| ["opam", "init", "--reinit", "--bare", "--no-setup", instance, url], | [ | ||||
| "opam", | |||||
| "init", | |||||
| "--reinit", | |||||
| "--bare", | |||||
| "--no-setup", | |||||
| "--root", | |||||
| self.opamroot, | |||||
| instance, | |||||
| url, | |||||
| ], | |||||
| env=self.env, | env=self.env, | ||||
| ) | ) | ||||
| def get_pages(self) -> Iterator[PageType]: | def get_pages(self) -> Iterator[PageType]: | ||||
| proc = Popen( | proc = Popen( | ||||
| [ | [ | ||||
| "opam", | "opam", | ||||
| "list", | "list", | ||||
| "--all", | "--all", | ||||
| "--no-switch", | "--no-switch", | ||||
| "--repos", | "--repos", | ||||
| self.instance, | self.instance, | ||||
| "--root", | |||||
| self.opamroot, | |||||
| "--normalise", | "--normalise", | ||||
| "--short", | "--short", | ||||
| ], | ], | ||||
| env=self.env, | env=self.env, | ||||
| stdout=PIPE, | stdout=PIPE, | ||||
| ) | ) | ||||
| if proc.stdout is not None: | if proc.stdout is not None: | ||||
| for line in io.TextIOWrapper(proc.stdout): | for line in io.TextIOWrapper(proc.stdout): | ||||
| yield line.rstrip("\n") | yield line.rstrip("\n") | ||||
| def get_origins_from_page(self, page: PageType) -> Iterator[ListedOrigin]: | def get_origins_from_page(self, page: PageType) -> Iterator[ListedOrigin]: | ||||
| """Convert a page of OpamLister repositories into a list of ListedOrigins""" | """Convert a page of OpamLister repositories into a list of ListedOrigins""" | ||||
| assert self.lister_obj.id is not None | assert self.lister_obj.id is not None | ||||
| # a page is just a package name | # a page is just a package name | ||||
| url = f"opam+{self.url}/packages/{page}/" | url = f"opam+{self.url}/packages/{page}/" | ||||
| yield ListedOrigin( | yield ListedOrigin( | ||||
| lister_id=self.lister_obj.id, | lister_id=self.lister_obj.id, | ||||
| visit_type="opam", | visit_type="opam", | ||||
| url=url, | url=url, | ||||
| last_update=None, | last_update=None, | ||||
| extra_loader_arguments={ | extra_loader_arguments={ | ||||
| "opam_root": self.env["OPAMROOT"], | "opam_root": self.opamroot, | ||||
| "opam_instance": self.instance, | "opam_instance": self.instance, | ||||
| "opam_url": self.url, | "opam_url": self.url, | ||||
| "opam_package": page, | "opam_package": page, | ||||
| }, | }, | ||||
| ) | ) | ||||