Page MenuHomeSoftware Heritage
Differential D5975 Diff 21703 swh/loader/package/opam/loader.py

Changeset View

Standalone View

View Options

swh/loader/package/opam/loader.py

  • This file was added.
# Copyright (C) 2021 The Software Heritage developers
vlorentzUnsubmitted
Done
Inline Actions

copyright header

vlorentz: copyright header
# 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 io
import os
from subprocess import PIPE, Popen, call
from typing import Iterator, List, Optional, Tuple
import attr
from swh.loader.package.loader import BasePackageInfo, PackageLoader
from swh.model.model import Person, Revision, RevisionType, Sha1Git
from swh.storage.interface import StorageInterface
@attr.s
class OpamPackageInfo(BasePackageInfo):
author = attr.ib(type=Person)
committer = attr.ib(type=Person)
version = attr.ib(type=str)
def opam_read(
cmd: List[str], init_error_msg_if_any: Optional[str] = None
ardumontUnsubmitted
Not Done
Inline Actions

As you did for the opam lister, can you please add a small explanation summary as a
first paragraph about what the loader is doing (ingesting one opam origin which may hold
multiple tarballs). Also, add a secondary paragrah explaining a bit the state
initialization process (adding the upstream url of the documentation is fine too [2]).

Thanks in advance.

[1] https://forge.softwareheritage.org/source/swh-lister/browse/master/swh/lister/opam/lister.py$0-26

[2] http://opam.ocaml.org/doc/Manual.html

ardumont: As you did for the opam lister, can you please add a small explanation summary as a first…
) -> Iterator[str]:
"""This executes and reads an opam command and yields the
output result one line at a time.
Args:
cmd: Opam command to execute as a list of string
init_error_msg_if_any: Error message to raise in case a problem occurs
during initialization
Raises:
ValueError with the init_error_msg_if_any content in case
stdout is not consumable (or something...) and the variable is provided.
Yields:
output line result of the command line
"""
with Popen(cmd, stdout=PIPE) as proc:
if proc.stdout is not None:
for line in io.TextIOWrapper(proc.stdout):
yield line
elif init_error_msg_if_any:
raise ValueError(init_error_msg_if_any)
class OpamLoader(PackageLoader[OpamPackageInfo]):
"""
Load all versions of a given package in a given opam repository.
The state of the opam repository is stored in a directory called an
opam root. Either the opam root has been created by the loader and we
simply re-use it, either it doesn't exist yet and we create it on the
first package we try to load (next packages will be able to re-use it).
ardumontUnsubmitted
Not Done
Inline Actions

Out of curiosity, do we really need to use the environment variable?
Can't we just use the --root flag parameter?

(not asking to change)

ardumont: Out of curiosity, do we really need to use the environment variable? Can't we just use the `…
aleoAuthorUnsubmitted
Done
Inline Actions

both are fine, the --root way will probably make the code easier to understand but I used the environment variable in the lister too... I'll change it here, don't you mind updating the lister ?

aleo: both are fine, the `--root` way will probably make the code easier to understand but I used the…
ardumontUnsubmitted
Not Done
Inline Actions

sure thing, fine for me.

ardumont: sure thing, fine for me.
Then we just ask the opam binary to give us the list of all versions of
the given package. For each version, we ask the opam binary to give us
the url to the tarball to archive.
"""
visit_type = "opam"
def __init__(
self,
storage: StorageInterface,
url: str,
opam_root: str,
opam_instance: str,
opam_url: str,
opam_package: str,
max_content_size: Optional[int] = None,
):
super().__init__(storage=storage, url=url, max_content_size=max_content_size)
self.opam_root = opam_root
self.opam_instance = opam_instance
self.opam_url = opam_url
self.opam_package = opam_package
if not os.path.isdir(opam_root):
if os.path.isfile(opam_root):
raise ValueError("invalid opam root")
else:
call(
[
"opam",
"init",
"--reinit",
"--bare",
"--no-setup",
"--root",
opam_root,
opam_instance,
opam_url,
]
)
elif not os.path.isfile(os.path.join(opam_root, "config")):
raise ValueError("invalid opam root")
ardumontUnsubmitted
Not Done
Inline Actions
raise ValueError("invalid opam root")
+ # ... # outside the class, at the top of the module before the class def
+ def opam_read(cmd: List[str], init_error_msg_if_any: Optional[str] = None) -> Iterator[str]:
+ """This executes and reads an opam command and yields the output result one line at a time.
+
+ Args:
+ cmd: Opam command to execute as a list of string
+ init_error_msg_if_any: Error message to raise in case a problem occurs
+ during initialization
+
+ Raises:
+ ValueError with the init_error_msg_if_any content in case
+ stdout is not consumable (or something...) and the variable is provided.
+
+ Yields:
+ output line result of the command line
+
+ """
+ with Popen(cmd, env=self.env, stdout=PIPE) as proc: # using this should be better as it closes what needs to
+ if proc.stdout is not None:
+ for line in io.TextIOWrapper(proc.stdout):
+ yield line
+ else if init_error_msg_if_any:
+ raise ValueError(init_error_msg_if_any)
+
def get_versions(self) -> List[str]:
- proc = Popen(
- [
- "opam",
- "show",
- "--color",
- "never",
- "--normalise",
- "-f",
- "all-versions",
- self.opam_package,
- ],
- env=self.env,
- stdout=PIPE,
- )
- if proc.stdout is not None:
- for line in io.TextIOWrapper(proc.stdout):
- # we only care about the first line
- # and there should be only one line anyway
- return line.split()
- raise ValueError(
- f"can't get versions for package {self.opam_package} \
- (at url {self.url}) from `opam show`"
- )
+ for line in opam_read([
+ "opam",
+ "show",
+ "--color",
+ "never",
+ "--normalise",
+ "-f",
+ "all-versions",
+ self.opam_package,
+ ], init_error_msg_if_any = (
+ f"can't get versions for package {self.opam_package} \
+ (at url {self.url}) from `opam show`"
+ ):
+ # only care about the first and only line which hold the versions information
+ # as a blank separated list
+ return line.split()
def get_default_version(self) -> str:

what do you think of the following?

That should decrease the repetition a bit.

ardumont: what do you think of the following? That should decrease the repetition a bit.
def get_versions(self) -> List[str]:
init_error_msg = f"can't get versions for package {self.opam_package} \
(at url {self.url}) from `opam show`"
for line in opam_read(
[
"opam",
"show",
"--color",
"never",
"--normalise",
"--root",
self.opam_root,
"-f",
"all-versions",
self.opam_package,
],
init_error_msg_if_any=init_error_msg,
):
# only care about the first and only line which hold the
# versions information as a blank separated list
return line.split()
raise ValueError(init_error_msg)
def get_default_version(self) -> str:
init_error_msg = f"can't get default version for package {self.opam_package} \
(at url {self.url}) from `opam show`"
for line in opam_read(
ardumontUnsubmitted
Not Done
Inline Actions
(at url {self.url}) from `opam show`"
)
def get_default_version(self) -> str:
- proc = Popen(
- [
- "opam",
- "show",
- "--color",
- "never",
- "--normalise",
- "-f",
- "version",
- self.opam_package,
- ],
- env=self.env,
- stdout=PIPE,
- )
- if proc.stdout is not None:
- for line in io.TextIOWrapper(proc.stdout):
- # we only care about the first element of the first line
- # and there should be only one element and one line anyway
- v = line.split()
- if len(v) == 1:
- return v[0]
- else:
- break
- raise ValueError(
- f"can't get default version for package {self.opam_package} \
- (at url {self.url}) from `opam show`"
- )
+ init_error_msg = (
+ f"can't get default version for package {self.opam_package} "
+ f"(at url {self.url}) from `opam show`"
+ )
+ for line in opam_read([
+ "opam",
+ "show",
+ "--color",
+ "never",
+ "--normalise",
+ "-f",
+ "version",
+ self.opam_package,
+ ], init_error_msg_if_any=init_error_msg):
+ # we only care about the first element of the first and only line
+ # which is the initial version of the package
+ versions = line.split()
+ if len(versions) != 1:
+ raise ValueError(init_error_msg)
+ return versions[0]
+
def get_enclosed_single_line_field(self, field, version) -> Optional[str]:

which then becomes according to the previous suggestion.

ardumont: which then becomes according to the previous suggestion.
[
"opam",
"show",
"--color",
"never",
"--normalise",
"--root",
self.opam_root,
"-f",
"version",
self.opam_package,
],
init_error_msg_if_any=init_error_msg,
):
# we only care about the first element of the first line
# and there should be only one element and one line anyway
v = line.split()
if len(v) != 1:
raise ValueError(init_error_msg)
return v[0]
raise ValueError(init_error_msg)
def get_enclosed_single_line_field(self, field, version) -> Optional[str]:
ardumontUnsubmitted
Not Done
Inline Actions
def get_enclosed_single_line_field(self, field, version) -> Optional[str]:
- proc = Popen(
+ for line in opam_read(
[
"opam",
"show",
"--color",
"never",
"--normalise",
"-f",
field,
f"{self.opam_package}.{version}",
- ],
- env=self.env,
- stdout=PIPE,
- )
- if proc.stdout is not None:
- for line in io.TextIOWrapper(proc.stdout):
- # we only care about the first line
- # and there should be only one line anyway
- # we also need to remove the enclosing " and the trailing \n
- return line[1:-2]
+ ]
+ ):
+ # we only care about the first and only line which needs to be
+ # cleaned up from the enclosing " and the trailing \n
+ return line[1:-2]
return None
def get_package_info(self, version: str) -> Iterator[Tuple[str, OpamPackageInfo]]:

finally ;)

ardumont: finally ;)
for line in opam_read(
[
"opam",
"show",
"--color",
"never",
vlorentzUnsubmitted
Not Done
Inline Actions

So this could potentiality contain multiple names per Person, right?

vlorentz: So this could potentiality contain multiple names per `Person`, right?
aleoAuthorUnsubmitted
Done
Inline Actions

Indeed. Is there another class that I should use then ?

aleo: Indeed. Is there another class that I should use then ?
vlorentzUnsubmitted
Not Done
Inline Actions

Unfortunately, the data-model doesn't allow this yet.

In the meantime, I wonder if we should use a hack like Github's to embed them in the message... https://docs.github.com/en/github/committing-changes-to-your-project/creating-and-editing-commits/creating-a-commit-with-multiple-authors

This probably needs to be discussed with the whole team

vlorentz: Unfortunately, the data-model doesn't allow this yet. In the meantime, I wonder if we should…
"--normalise",
"--root",
self.opam_root,
"-f",
field,
f"{self.opam_package}.{version}",
]
):
# we only care about the first line
# and there should be only one line anyway
# we also need to remove the enclosing " and the trailing \n
return line[1:-2]
vlorentzUnsubmitted
Not Done
Inline Actions

if there are no release notes, could you write the version number as message?

vlorentz: if there are no release notes, could you write the version number as message?
aleoAuthorUnsubmitted
Done
Inline Actions

Done !

aleo: Done !
return None
vlorentzUnsubmitted
Not Done
Inline Actions

Oh :(

I see there is a publication date on Opam's web interface, is there no way we could fetch that?

vlorentz: Oh :( I see there is a publication date on Opam's web interface, is there no way we could…
aleoAuthorUnsubmitted
Done
Inline Actions

I don't see an easy way to do that. It's not really part of the opam repository. I guess you can get it if the opam repository is a git repository by checking the git history but I'm not sure we want to do that neither if it always makes sense.

Especially it won't be possible in the case of a local opam repository (used for tests) or for the Coq opam repository which is there: https://coq.inria.fr/opam/released/

The two opam developpers are on vacation, I'll ask them if there's an easy way to do it but I don't expect so.

aleo: I don't see an easy way to do that. It's not really part of the opam repository. I guess you…
def get_package_info(self, version: str) -> Iterator[Tuple[str, OpamPackageInfo]]:
branch_name = f"{self.opam_package}.{version}"
url = self.get_enclosed_single_line_field("url.src:", version)
if url is None:
raise ValueError(
f"can't get field url.src: for version {version} of package {self.opam_package} \
(at url {self.url}) from `opam show`"
)
authors_field = self.get_enclosed_single_line_field("authors:", version)
fullname = b"" if authors_field is None else str.encode(authors_field)
author = Person(fullname=fullname, name=None, email=None)
maintainer_field = self.get_enclosed_single_line_field("maintainer:", version)
fullname = b"" if maintainer_field is None else str.encode(maintainer_field)
committer = Person(fullname=fullname, name=None, email=None)
yield branch_name, OpamPackageInfo(
url=url, filename=None, author=author, committer=committer, version=version
)
def build_revision(
self, p_info: OpamPackageInfo, uncompressed_path: str, directory: Sha1Git
) -> Optional[Revision]:
return Revision(
type=RevisionType.TAR,
author=p_info.author,
committer=p_info.committer,
message=str.encode(p_info.version),
date=None,
committer_date=None,
parents=(),
directory=directory,
synthetic=True,
)
About · Developer information · Legal