Differential D6009 Diff 21710 swh/loader/package/opam/loader.py

Changeset View

Standalone View

swh/loader/package/opam/loader.py

Show All 18 Lines
class OpamPackageInfo(BasePackageInfo):		class OpamPackageInfo(BasePackageInfo):
author = attr.ib(type=Person)		author = attr.ib(type=Person)
committer = attr.ib(type=Person)		committer = attr.ib(type=Person)
version = attr.ib(type=str)		version = attr.ib(type=str)


def opam_read(		def opam_read(
cmd: List[str], init_error_msg_if_any: Optional[str] = None		cmd: List[str], init_error_msg_if_any: Optional[str] = None
) -> Iterator[str]:		) -> Optional[str]:
"""This executes and reads an opam command and yields the		"""This executes an opam command and returns the first line of the output.
output result one line at a time.

Args:		Args:
cmd: Opam command to execute as a list of string		cmd: Opam command to execute as a list of string
init_error_msg_if_any: Error message to raise in case a problem occurs		init_error_msg_if_any: Error message to raise in case a problem occurs
during initialization		during initialization

Raises:		Raises:
ValueError with the init_error_msg_if_any content in case		ValueError with the init_error_msg_if_any content in case stdout is not
stdout is not consumable (or something...) and the variable is provided.		consumable and the variable is provided with non empty value.

Yields:		Returns:
output line result of the command line		the first line of the executed command output

"""		"""
with Popen(cmd, stdout=PIPE) as proc:		with Popen(cmd, stdout=PIPE) as proc:
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		# care only for the first line output result (mostly blank separated
		# values, callers will deal with the parsing of the line)
		return line
elif init_error_msg_if_any:		elif init_error_msg_if_any:
raise ValueError(init_error_msg_if_any)		raise ValueError(init_error_msg_if_any)
		return None


class OpamLoader(PackageLoader[OpamPackageInfo]):		class OpamLoader(PackageLoader[OpamPackageInfo]):
"""		"""
Load all versions of a given package in a given opam repository.		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		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		opam root. Either the opam root has been created by the loader and we
Show All 40 Lines	):
opam_instance,		opam_instance,
opam_url,		opam_url,
]		]
)		)
elif not os.path.isfile(os.path.join(opam_root, "config")):		elif not os.path.isfile(os.path.join(opam_root, "config")):
raise ValueError("invalid opam root")		raise ValueError("invalid opam root")

def get_versions(self) -> List[str]:		def get_versions(self) -> List[str]:
init_error_msg = f"can't get versions for package {self.opam_package} \		versions = opam_read(
(at url {self.url}) from `opam show`"
for line in opam_read(
[		[
"opam",		"opam",
"show",		"show",
"--color",		"--color",
"never",		"never",
"--normalise",		"--normalise",
"--root",		"--root",
self.opam_root,		self.opam_root,
"-f",		"-f",
"all-versions",		"all-versions",
self.opam_package,		self.opam_package,
],		],
init_error_msg_if_any=init_error_msg,		init_error_msg_if_any=(
):		f"can't get versions for package {self.opam_package} "
# only care about the first and only line which hold the		f"(at url {self.url}) from `opam show`"
# versions information as a blank separated list		),
return line.split()		)
raise ValueError(init_error_msg)		return versions.split() if versions else []

def get_default_version(self) -> str:		def get_default_version(self) -> str:

init_error_msg = f"can't get default version for package {self.opam_package} \		init_error_msg = f"can't get default version for package {self.opam_package} \
(at url {self.url}) from `opam show`"		(at url {self.url}) from `opam show`"
for line in opam_read(		# we only care about the first element of the first line
		# which is the initial version
		versions_ = opam_read(
[		[
"opam",		"opam",
"show",		"show",
"--color",		"--color",
"never",		"never",
"--normalise",		"--normalise",
"--root",		"--root",
self.opam_root,		self.opam_root,
"-f",		"-f",
"version",		"version",
self.opam_package,		self.opam_package,
],		],
init_error_msg_if_any=init_error_msg,		init_error_msg_if_any=init_error_msg,
):		)
# we only care about the first element of the first line		if not versions_:
# and there should be only one element and one line anyway
v = line.split()
if len(v) != 1:
raise ValueError(init_error_msg)		raise ValueError(init_error_msg)
return v[0]		versions = versions_.split()
		if len(versions) != 1:
raise ValueError(init_error_msg)		raise ValueError(init_error_msg)
		return versions[0]

def get_enclosed_single_line_field(self, field, version) -> Optional[str]:		def get_enclosed_single_line_field(self, field, version) -> Optional[str]:
for line in opam_read(		result = opam_read(
[		[
"opam",		"opam",
"show",		"show",
"--color",		"--color",
"never",		"never",
"--normalise",		"--normalise",
"--root",		"--root",
self.opam_root,		self.opam_root,
"-f",		"-f",
field,		field,
f"{self.opam_package}.{version}",		f"{self.opam_package}.{version}",
]		]
):		)
# we only care about the first line
# and there should be only one line anyway		# this needs to be cleaned up a bit (remove enclosing " and the trailing \n)
# we also need to remove the enclosing " and the trailing \n		return result[1:-2] if result else None
return line[1:-2]
return None

def get_package_info(self, version: str) -> Iterator[Tuple[str, OpamPackageInfo]]:		def get_package_info(self, version: str) -> Iterator[Tuple[str, OpamPackageInfo]]:

branch_name = f"{self.opam_package}.{version}"		branch_name = f"{self.opam_package}.{version}"
url = self.get_enclosed_single_line_field("url.src:", version)		url = self.get_enclosed_single_line_field("url.src:", version)

if url is None:		if url is None:
raise ValueError(		raise ValueError(
Show All 31 Lines