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

Changeset View

Standalone View

View Options

swh/loader/package/opam/loader.py

  • This file was added.
import io
vlorentzUnsubmitted
Done
Inline Actions

copyright header

vlorentz: copyright header
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)
class OpamLoader(PackageLoader[OpamPackageInfo]):
visit_type = "opam"
def __init__(
self,
storage: StorageInterface,
url: str,
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…
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
self.env = os.environ.copy()
self.env["OPAMROOT"] = opam_root
if not os.path.isdir(opam_root):
if os.path.isfile(opam_root):
print("invalid opam root")
exit(1)
else:
call(
[
"opam",
"init",
"--reinit",
"--bare",
"--no-setup",
opam_instance,
opam_url,
],
env=self.env,
)
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.
elif not os.path.isfile(os.path.join(opam_root, "config")):
print("invalid opam root")
exit(1)
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()
print(
f"can't get versions for package {self.opam_package} \
(at url {self.url})"
)
exit(1)
def get_default_version(self) -> str:
proc = Popen(
[
"opam",
"show",
"--color",
"never",
"--normalise",
"-f",
"version",
self.opam_package,
],
env=self.env,
stdout=PIPE,
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.
)
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
print(
f"can't get default version for package {self.opam_package} \
(at url {self.url})"
)
exit(1)
def get_enclosed_single_line_field(self, field, version) -> Optional[str]:
proc = Popen(
[
"opam",
"show",
"--color",
"never",
"--normalise",
"-f",
field,
f"{self.opam_package}.{version}",
],
env=self.env,
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.
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]
return None
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:
print(
f"can't get field url.src: for version {version} of package {self.opam_package} \
(at url {self.url})"
)
exit(1)
authors_field = self.get_enclosed_single_line_field("authors:", version)
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 ;)
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)
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…
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,
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 !
committer=p_info.committer,
message=str.encode(p_info.version),
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…
date=None,
committer_date=None,
parents=(),
directory=directory,
synthetic=True,
)
About · Developer information · Legal