Source code for swh.lister.opam.lister

# Copyright (C) 2021-2023  The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information

import logging
import os
import shutil
from subprocess import PIPE, run
from typing import Any, Dict, Iterator, Optional

from swh.lister.pattern import StatelessLister
from swh.scheduler.interface import SchedulerInterface
from swh.scheduler.model import ListedOrigin

from ..pattern import CredentialsType

logger = logging.getLogger(__name__)

PageType = str

[docs] def opam() -> str: """Get the path to the opam executable. Raises: EnvironmentError if no opam executable is found """ ret = shutil.which("opam") if not ret: raise EnvironmentError("No opam executable found in path {os.environ['PATH']}") return ret
[docs] class OpamLister(StatelessLister[PageType]): """ List all repositories hosted on an opam repository. On initialisation, we create an opam root, with no ocaml compiler (no switch) as we won't need it and it's costly. In this opam root, we add a single opam repository (url) and give it a name (instance). Then, to get pages, we just ask opam to list all the packages for our opam repository in our opam root. Args: url: base URL of an opam repository (for instance instance: string identifier for the listed repository """ # Part of the lister API, that identifies this lister LISTER_NAME = "opam" def __init__( self, scheduler: SchedulerInterface, url: str, instance: Optional[str] = None, credentials: CredentialsType = None, max_origins_per_page: Optional[int] = None, max_pages: Optional[int] = None, enable_origins: bool = True, opam_root: str = "/tmp/opam/", ): super().__init__( scheduler=scheduler, credentials=credentials, url=url, instance=instance, max_origins_per_page=max_origins_per_page, max_pages=max_pages, enable_origins=enable_origins, ) self.env = os.environ.copy() # Opam root folder is initialized in the :meth:`get_pages` method as no # side-effect should happen in the constructor to ease instantiation self.opam_root = opam_root
[docs] def get_pages(self) -> Iterator[PageType]: # Initialize the opam root directory opam_init(self.opam_root, self.instance, self.url, self.env) # Actually list opam instance data proc = run( [ opam(), "list", "--all", "--no-switch", "--safe", "--repos", self.instance, "--root", self.opam_root, "--normalise", "--short", ], env=self.env, stdout=PIPE, text=True, check=True, ) if proc.stdout is not None: yield from proc.stdout.splitlines()
[docs] def get_origins_from_page(self, page: PageType) -> Iterator[ListedOrigin]: """Convert a page of OpamLister repositories into a list of ListedOrigins""" assert is not None # a page is just a package name url = f"opam+{self.url}/packages/{page}/" yield ListedOrigin(, visit_type="opam", url=url, last_update=None, extra_loader_arguments={ "opam_root": self.opam_root, "opam_instance": self.instance, "opam_url": self.url, "opam_package": page, }, )
[docs] def opam_init(opam_root: str, instance: str, url: str, env: Dict[str, Any]) -> None: """Initialize an opam_root folder. Args: opam_root: The opam root folder to initialize instance: Name of the opam repository to add or initialize url: The associated url of the opam repository to add or initialize env: The global environment to use for the opam command. Returns: None. """ if not os.path.exists(opam_root) or not os.listdir(opam_root): command = [ opam(), "init", "--reinit", "--bare", "--no-setup", "--root", opam_root, instance, url, ] else: # The repository exists and is populated, we just add another instance in the # repository. If it's already setup, it's a noop command = [ opam(), "repository", "add", "--set-default", "--root", opam_root, instance, url, ] # Actually execute the command run(command, env=env, check=True)