Skip to content

kraken.std.dist

kraken.std.dist

Create distributions (archives) from build artifacts and resources.

ArchiveWriter

Bases: ABC

Base class to write an archive file.

Source code in kraken/std/dist.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
class ArchiveWriter(abc.ABC):
    """Base class to write an archive file."""

    @abc.abstractmethod
    def add_file(self, arcname: str, path: Path) -> None:
        """Add a file to the archive."""

    @abc.abstractmethod
    def close(self) -> None:
        pass

    def add_path(self, arcname: str, path: Path) -> None:
        """Recursively add the contents of all files under *path*."""

        if path.is_dir():
            for item in path.iterdir():
                self.add_path(arcname + "/" + item.name, item)
        else:
            self.add_file(arcname, path)

    def __enter__(self) -> ArchiveWriter:
        return self

    def __exit__(self, *a: Any) -> None:
        self.close()
add_file abstractmethod 
add_file(arcname: str, path: Path) -> None

Add a file to the archive.

Source code in kraken/std/dist.py 
151
152
153
@abc.abstractmethod
def add_file(self, arcname: str, path: Path) -> None:
    """Add a file to the archive."""
add_path 
add_path(arcname: str, path: Path) -> None

Recursively add the contents of all files under path.

Source code in kraken/std/dist.py 
159
160
161
162
163
164
165
166
def add_path(self, arcname: str, path: Path) -> None:
    """Recursively add the contents of all files under *path*."""

    if path.is_dir():
        for item in path.iterdir():
            self.add_path(arcname + "/" + item.name, item)
    else:
        self.add_file(arcname, path)

DistributionTask

Bases: Task

Create an archive from a set of files and resources.

Source code in kraken/std/dist.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
class DistributionTask(Task):
    """Create an archive from a set of files and resources."""

    #: The output filename. If a string is specified, that string is treated relative
    #: to the project output directory. A Path object is treated relative to the project
    #: directory. Unless the :attr:`archive_type` property is set, the suffix will determine
    #: the type of archive that is created.
    output_file: Property[Path]

    #: The type of archive that will be created. Can be zip, tgz and tar.
    archive_type: Property[str]

    #: Prefix to add to all files added to the distribution.
    prefix: Property[str]

    #: A list of resources to include.
    resources: Property[list[ConfiguredResource]] = Property.default_factory(list)

    #: A resource that describes the output file.
    _output_file_resource: Property[Resource] = Property.output()

    def __init__(self, name: str, project: Project) -> None:
        super().__init__(name, project)
        self._output_file_resource.set(self.output_file.map(lambda p: Resource("dist", p)))

    # Task

    def execute(self) -> None:
        output_file = self.output_file.get()
        archive_type = self.archive_type.get_or(None)
        if archive_type is None:
            archive_type = output_file.suffix.lstrip(".")
            archive_type = {"tgz": "tar.gz", "txz": "tar.xz", "tbz2": "tar.bz2"}.get(archive_type, archive_type)
        assert isinstance(archive_type, str)

        print("Writing archive", colored(str(output_file), "yellow"))
        output_file.parent.mkdir(parents=True, exist_ok=True)
        with wopen_archive(output_file, archive_type) as archive:
            for resource in self.resources.get():
                arcname = resource.options.arcname
                if arcname is None:
                    arcname = str(resource.path.relative_to(self.project.directory))
                print(
                    "  +",
                    colored(arcname or ".", "green"),
                    f"({resource.path})" if arcname != str(resource.path) else "",
                )
                add_to_archive(
                    archive,
                    arcname,
                    self.project.directory / resource.path,
                    resource.path,
                    resource.options.exclude,
                    resource.options.include,
                )

add_to_archive 

add_to_archive(
    writer: ArchiveWriter,
    arcname: str,
    path: Path,
    test_path: Path | None = None,
    exclude: Sequence[str] = (),
    include: Sequence[str] | None = None,
) -> None

Recursively adds path to the archive writer under consideration of the exclude and include glob patterns that are tested against test_path.

The glob patterns for the exclude and include arguments are tested against the full relative test_path as well as the individual filename. (NOTE(niklas.rosenstein): To support full .gitignore like behaviour, we'd need to test the entire range from name to full path).

:param writer: The archive writer implementation. :param arcname: The name to give path. Sub-paths are appended to this name as normal. :param path: The path to write to the archive. :param test_path: The path to test the exclude/include patterns against. Sub-paths are appended to this as normal. Defaults to path. :param exclude: A sequence of glob patterns that, if matching, cause an element (file or directory) to be excluded. :param include: If specified, must be a sequence of glob patterns that will cause a file or directory to only be added if any pattern matches, and no exclude pattern matches.

Source code in kraken/std/dist.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def add_to_archive(
    writer: ArchiveWriter,
    arcname: str,
    path: Path,
    test_path: Path | None = None,
    exclude: Sequence[str] = (),
    include: Sequence[str] | None = None,
) -> None:
    """Recursively adds *path* to the archive *writer* under consideration of the *exclude* and *include* glob
    patterns that are tested against *test_path*.

    The glob patterns for the *exclude* and *include* arguments are tested against the full relative *test_path*
    as well as the individual filename. (NOTE(niklas.rosenstein): To support full .gitignore like behaviour, we'd
    need to test the entire range from name to full path).

    :param writer: The archive writer implementation.
    :param arcname: The name to give *path*. Sub-paths are appended to this name as normal.
    :param path: The path to write to the archive.
    :param test_path: The path to test the exclude/include patterns against. Sub-paths are appended to this as normal.
        Defaults to *path*.
    :param exclude: A sequence of glob patterns that, if matching, cause an element (file or directory) to be excluded.
    :param include: If specified, must be a sequence of glob patterns that will cause a file or directory to only be
        added if any pattern matches, and no *exclude* pattern matches.
    """

    test_path = test_path or path
    s_test_path = str(test_path)

    if any(fnmatch(s_test_path, x) or fnmatch(test_path.name, x) for x in exclude):
        return
    if include is not None and not any(fnmatch(s_test_path, x) or fnmatch(test_path.name, x) for x in include):
        return

    if path.is_dir():
        for item in path.iterdir():
            add_to_archive(writer, arcname + "/" + item.name, item, test_path / item.name, exclude, include)
    else:
        writer.add_file(arcname, path)

dist 

dist(
    *,
    name: str,
    dependencies: (
        Sequence[str | Task]
        | Mapping[
            str, Mapping[str, Any] | IndividualDistOptions
        ]
    ),
    output_file: str | Path,
    archive_type: str | None = None,
    prefix: str | None = None,
    project: Project | None = None
) -> DistributionTask

Create a task that produces a distribution from the resources provided by the tasks specified with include.

:param name: The name of the task. :param dependencies: A list of tasks or task selectors (resolved relative to the project) that provide the resources that should be included in the generated archive. If a dictionary is specified, instead, it must map each dependency to a dictionary of settings that can be deserialized into :class:IndividualDistOptions. :param output_file: The output filename to write to. If a string is specified, it will be treated relative to the project build directory. If a path object is specified, it will be treated relative to the project directory. :param archive_type: The type of archive to create (e.g. zip, tar, tar.gz, tar.bz2 or tar.xz). If left empty, the archive type is derived from the output_filename suffix. :param project: The project to create the task for.

Source code in kraken/std/dist.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def dist(
    *,
    name: str,
    dependencies: Sequence[str | Task] | Mapping[str, Mapping[str, Any] | IndividualDistOptions],
    output_file: str | Path,
    archive_type: str | None = None,
    prefix: str | None = None,
    project: Project | None = None,
) -> DistributionTask:
    """Create a task that produces a distribution from the resources provided by the tasks specified with *include*.

    :param name: The name of the task.
    :param dependencies: A list of tasks or task selectors (resolved relative to the *project*) that provide the
        resources that should be included in the generated archive. If a dictionary is specified, instead, it must
        map each dependency to a dictionary of settings that can be deserialized into :class:`IndividualDistOptions`.
    :param output_file: The output filename to write to. If a string is specified, it will be treated relative to the
        project build directory. If a path object is specified, it will be treated relative to the project directory.
    :param archive_type: The type of archive to create (e.g. `zip`, `tar`, `tar.gz`, `tar.bz2` or `tar.xz`). If left
        empty, the archive type is derived from the *output_filename* suffix.
    :param project: The project to create the task for.
    """

    project = project or Project.current()

    if isinstance(output_file, str):
        output_file = project.build_directory / output_file
    else:
        output_file = project.directory / output_file

    if isinstance(dependencies, Sequence):
        dependencies = cast(
            Mapping[str, Union[Mapping[str, Any], IndividualDistOptions]], {d: {} for d in dependencies}
        )
    dependencies_map = {
        k: databind.json.load(v, IndividualDistOptions) if not isinstance(v, IndividualDistOptions) else v
        for k, v in dependencies.items()
    }
    dependencies_set = project.resolve_tasks(dependencies_map)

    # This associates the IndividualDistOptions specified in *dependencies* to the Resource(s)
    # provided by the task(s).
    resources = (
        dependencies_set.select(Resource)
        .dict_supplier()
        .map(
            lambda resources: get_configured_resources(
                resources,
                dependencies_map,
                dependencies_set,
            )
        )
    )

    dist_task = project.task(name, DistributionTask)
    dist_task.resources = resources
    dist_task.output_file = output_file
    dist_task.archive_type = archive_type
    dist_task.prefix = prefix
    return dist_task

wopen_archive 

wopen_archive(path: Path, type_: str) -> ArchiveWriter

Open an archive at path for writing. The type_ indicates what type of archive will be created. Accepted values for type_ are zip, tar, tar.gz, tar.bz2 and tar.xz.

Source code in kraken/std/dist.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def wopen_archive(path: Path, type_: str) -> ArchiveWriter:
    """Open an archive at *path* for writing. The *type_* indicates what type of archive will be created.
    Accepted values for *type_* are `zip`, `tar`, `tar.gz`, `tar.bz2` and `tar.xz`."""

    if type_.startswith("tar."):
        return TarArchiveWriter(path, cast(Any, type_.partition(".")[-1]))
    elif type_ == "tar":
        return TarArchiveWriter(path, "")
    elif type_ == "zip":
        return ZipArchiveWriter(path)
    else:
        raise ValueError(f"unsupported archive type: {type_!r}")