Verified Commit 4e4d5bbe authored by David Runge's avatar David Runge 🐿
Browse files

Add iterator over repository database

Add the public method `db_file_as_models()`, which yields all members of
a repository database as a Tuple of the pkgbase name and the pkgbase
data (represented as an instance of models.OutputPackageBase).

Add a test for `db_file_as_models()`.
parent 53f6c29d
from pathlib import Path
from typing import Dict, Iterator, Tuple
from repo_management import convert, defaults, files, models
def db_file_as_models(db_path: Path, compression: str = "gz") -> Iterator[Tuple[str, models.OutputPackageBase]]:
"""Read a repository database and yield the name of each pkgbase and the respective data (represented as an instance
of models.OutputPackageBase) in a Tuple.
db_path: Path
A Path representing a repository database file
compression: str
The compression used for the repository database file (support depends on tarfile). Defaults to "gz" (gzip
Iterator[Tuple[str, models.OutputPackageBase]]:
A Tuple holding the name of a pkgbase and its accompanying data in an instance of models.OutputPackageBase
packages: Dict[str, models.OutputPackageBase] = {}
package_descs: Dict[str, models.PackageDesc] = {}
package_files: Dict[str, models.Files] = {}
for member in files._db_file_member_as_model(db_file=files._read_db_file(db_path=db_path, compression=compression)):
if member.member_type == defaults.RepoDbMemberType.DESC:
package_descs.update({ convert._desc_data_to_model(})
if member.member_type == defaults.RepoDbMemberType.FILES:
package_files.update({ convert._files_data_to_model(})
for (name, package_desc) in package_descs.items():
if packages.get(package_desc.base):
packages[package_desc.base].packages += [
convert._transform_package_desc_to_output_package(desc=package_desc, files=package_files.get(name))
package_desc.base: models.OutputPackageBase(
desc=package_desc, files=package_files.get(name)
for (name, package) in packages.items():
yield (name, package)
import os
from pathlib import Path
from typing import Iterator
from pytest import fixture
from repo_management import models, operations
from .fixtures import create_db_file
def create_gz_db_file() -> Iterator[Path]:
db_file = create_db_file()
yield db_file
def test_db_file_as_models(create_gz_db_file: Path) -> None:
for (name, model) in operations.db_file_as_models(db_path=create_gz_db_file):
assert isinstance(name, str)
assert isinstance(model, models.OutputPackageBase)
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment