operations.py 4.19 KB
Newer Older
1
from os.path import join
2
from pathlib import Path
3
from typing import AsyncIterator, Dict, Tuple
4

5
import aiofiles
6
7
import orjson

8
9
10
from repo_management import convert, defaults, files, models


11
12
13
async def db_file_as_models(
    db_path: Path, compression: str = "gz"
) -> AsyncIterator[Tuple[str, models.OutputPackageBase]]:
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
    """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.

    Parameters
    ----------
    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
        compression)

    Returns
    -------
    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] = {}
34
    async for member in files._db_file_member_as_model(  # pragma: no cover
35
36
        db_file=await files._read_db_file(db_path=db_path, compression=compression)
    ):
37
        if member.member_type == defaults.RepoDbMemberType.DESC:
38
            package_descs.update({member.name: await convert._desc_data_to_model(member.data)})
39
        if member.member_type == defaults.RepoDbMemberType.FILES:
40
            package_files.update({member.name: await convert._files_data_to_model(member.data)})
41
42
43

    for (name, package_desc) in package_descs.items():
        if packages.get(package_desc.base):
44
            packages[package_desc.base].packages += [package_desc.get_output_package(files=package_files.get(name))]
45
46
47
48
        else:
            packages.update(
                {
                    package_desc.base: models.OutputPackageBase(
49
                        base=package_desc.base,
50
51
52
                        makedepends=package_desc.makedepends,
                        packager=package_desc.packager,
                        version=package_desc.version,
53
                        packages=[package_desc.get_output_package(files=package_files.get(name))],
54
55
56
57
58
59
                    )
                }
            )

    for (name, package) in packages.items():
        yield (name, package)
60
61


62
async def dump_db_to_json_files(input_path: Path, output_path: Path) -> None:
63
64
65
66
67
68
69
70
71
72
73
    """Read a repository database file and dump each pkgbase contained in it to a separate JSON file below a defined
    output directory

    Parameters
    ----------
    input_path: Path
        The input file to read and parse
    output_path: Path
        A directory in which to
    """

74
75
76
    async for name, model in db_file_as_models(db_path=input_path):
        async with aiofiles.open(join(output_path, f"{name}.json"), "wb") as output_file:
            await output_file.write(
77
78
79
80
                orjson.dumps(
                    model.dict(), option=orjson.OPT_INDENT_2 | orjson.OPT_APPEND_NEWLINE | orjson.OPT_SORT_KEYS
                )
            )
81
82


83
async def create_db_from_json_files(
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
    input_path: Path, output_path: Path, db_type: defaults.RepoDbType = defaults.RepoDbType.DEFAULT
) -> None:
    """Create a repository database from a list of JSON files found in a directory

    Parameters
    ----------
    input_path: Path
        A directory from which to read JSON files
    output_path: Path
        A file to which to write a repository database
    db_type: defaults.RepoDbType
        A member of the defaults.RepoDbType IntEnum to define what type of repository database to create:
        Either defaults.RepoDbType.DEFAULT for the default .db database or defaults.RepoDbType.FILES for the .files
        database (defaults to defaults.RepoDbType.DEFAULT)
    """

    repodbfile = convert.RepoDbFile()
101
    database = await files._write_db_file(path=output_path)
102
    async for path in files._json_files_in_directory(path=input_path):  # pragma: no cover
103
104
        model = await files._read_pkgbase_json_file(path)
        await files._stream_package_base_to_db(
105
106
107
108
109
            db=database,
            model=model,
            repodbfile=repodbfile,
            db_type=db_type,
        )