Skip to content

phoenixpackagecleanup.clean_channel

Channel interaction module.

Functions:

Name Description
clean_up_channel

Clean up the package channel by sending DELETE requests to the delete api of the channel.
download_file

Download a file (similar to wget) using urllib.
get_argument_parser

Get phoenix package clean-up argument parser.
load_repodata

Load the repodata.json file.
main

Phoenix Package Clean-up entrypoint.
remove_version_matching_regex

Remove packages matching a regex from the package collection.
select_package_to_delete

Select the packages to clean-up from a package collection based on rules inspired by gitlab's clean-up rules.

clean_up_channel 

clean_up_channel(packages_to_delete, delete_api_url, api_token, timeout_s)

Clean up the package channel by sending DELETE requests to the delete api of the channel.

Parameters:

Name Type Description Default
packages_to_delete PackageInfoCollection

Collections of package to delete in the channel. The packages filenames will be appended to the delete_api_url to create the DELETE request.

 required
delete_api_url str

API url to delete the packages. Eg: "https://prefix.dev/api/v1/delete/phoenix-dev/linux-64/"

 required
api_token str

Token to identify with the delete API.

 required
timeout_s int

timeout in seconds for the delete request.

 required
Source code in src/phoenixpackagecleanup/clean_channel.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def clean_up_channel(
    packages_to_delete: PackageInfoCollection, delete_api_url: str, api_token: str, timeout_s: int
) -> None:
    """Clean up the package channel by sending DELETE requests to the delete api of the channel.

    Parameters
    ----------
    packages_to_delete : PackageInfoCollection
        Collections of package to delete in the channel. The packages filenames will be appended to the
        `delete_api_url` to create the DELETE request.
    delete_api_url : str
        API url to delete the packages. Eg: "https://prefix.dev/api/v1/delete/phoenix-dev/linux-64/"
    api_token : str
        Token to identify with the delete API.
    timeout_s : int
        timeout in seconds for the delete request.
    """
    for package_info_list in packages_to_delete.packages_info.values():
        for package_info in package_info_list:
            delete_url = delete_api_url + f"/{package_info.filename}"

            delete_request = urllib.request.Request(  # noqa: S310 delete url is user's responsability (from configuration)
                url=delete_url, headers={"Authorization": f"Bearer {api_token}"}, method="DELETE"
            )
            try:
                _ = urllib.request.urlopen(delete_request, timeout=timeout_s)  # noqa: S310 delete url is user's responsability (from configuration)
            except HTTPError:
                logging.getLogger(__name__).warning("Failed delete request with delete_url: %s", delete_url)
                raise

download_file 

download_file(url, filepath, user_agent, timeout_s)

Download a file (similar to wget) using urllib.

The main reason we need this method is to download the repodata.json file. wget works out of the box, but when using urllib, we have to specify a User-agent otherwise the request is rejected (403: Forbidden)

Parameters:

Name Type Description Default
url str

URL of the file to download

 required
filepath Path

Path to where the downloaded data will be written

 required
user_agent str

User-agent value to use in the request header

 required
timeout_s int

Timeout in second for the request

 required
Source code in src/phoenixpackagecleanup/clean_channel.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def download_file(url: str, filepath: Path, user_agent: str, timeout_s: int) -> None:
    """Download a file (similar to wget) using urllib.

    The main reason we need this method is to download the repodata.json file. wget works out of the box, but
    when using urllib, we have to specify a User-agent otherwise the request is rejected (403: Forbidden)

    Parameters
    ----------
    url : str
        URL of the file to download
    filepath : Path
        Path to where the downloaded data will be written
    user_agent : str
        User-agent value to use in the request header
    timeout_s : int
        Timeout in second for the request
    """
    file_request = urllib.request.Request(url=url, headers={"User-agent": user_agent})  # noqa: S310 Allow any file in general purpose function
    # if we get an error from server, it will be raised by urlopen, no need to catch
    response = urllib.request.urlopen(file_request, timeout=timeout_s)  # noqa: S310 Allow any file in general purpose function
    with filepath.open("wb") as of:
        of.write(response.read())

get_argument_parser 

get_argument_parser()

Get phoenix package clean-up argument parser.

Source code in src/phoenixpackagecleanup/clean_channel.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
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
def get_argument_parser() -> argparse.ArgumentParser:
    """Get phoenix package clean-up argument parser."""
    parser = argparse.ArgumentParser(
        description="Program which deletes unused packages", formatter_class=argparse.ArgumentDefaultsHelpFormatter
    )
    parser.add_argument("-t", "--token", help="Token to be used to log in prefix.dev", required=True, type=str)
    parser.add_argument(
        "-c",
        "--channel_url",
        help="Base URL of the channel, for instance https://prefix.dev/phoenix-dev/. It is combined with the "
        "channel's subdirectories to get a working url.",
        required=True,
        type=str,
    )
    parser.add_argument(
        "-a",
        "--api_delete_url",
        help="URL to submit DELETE requests for packages to clean-up, for instance "
        "https://prefix.dev/api/v1/delete/phoenix-dev/linux-64. It is combined with the channel's subdirectories"
        " to get a working url for API calls.",
        required=True,
        type=str,
    )
    parser.add_argument(
        "-n",
        "--min_number_of_packages",
        help="Minimum number of packages version to keep for a package",
        required=True,
        type=int,
    )
    parser.add_argument(
        "-d",
        "--min_age_days",
        help="Token to be used to define number of days to keep packages",
        required=True,
        type=int,
    )
    parser.add_argument(
        "-b",
        "--keep_version_regex",
        help="Packages which are not matching this regex will be considered for deletion. ",
        required=True,
        type=str,
    )
    parser.add_argument(
        "-l", "--logfile", help="Path to where the scripts log should be written", required=True, type=Path
    )
    parser.add_argument(
        "-v",
        "--loglevel",
        type=LOGGING_LEVELS,
        action=EnumAction,
        dest="logging_level",
        help="Log level as defined by the python logging library",
        required=True,
        default=LOGGING_LEVELS.WARNING,
    )
    return parser

load_repodata 

load_repodata(repodata_filepath)

Load the repodata.json file.

Parameters:

Name Type Description Default
repodata_filepath Path

Path to the repodata.json file

 required

Returns:

Type Description
ChannelPackages

Packages in the channel, grouped by package name
Source code in src/phoenixpackagecleanup/clean_channel.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def load_repodata(repodata_filepath: Path) -> PackageInfoCollection:
    """Load the repodata.json file.

    Parameters
    ----------
    repodata_filepath : Path
        Path to the repodata.json file

    Returns
    -------
    ChannelPackages
        Packages in the channel, grouped by package name
    """
    with repodata_filepath.open("r") as file:
        repodata = json.load(file)
    return parse_repodata(repodata)

main 

main()

Phoenix Package Clean-up entrypoint.

Source code in src/phoenixpackagecleanup/clean_channel.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
def main() -> None:
    """Phoenix Package Clean-up entrypoint."""
    parser = get_argument_parser()
    args = parser.parse_args()

    init_logging("PhoenixPackageCleanUp", args.logfile, args.logging_level)
    logger = logging.getLogger(__name__)

    channel_subdirs = ["noarch", "linux-64", "linux-aarch64", "linux-ppc64le", "osx-64", "osx-arm64", "win-64"]
    failed_all_subdir = True

    for subdir in channel_subdirs:
        # https://prefix.dev/phoenix-dev/linux-64/repodata.json
        # https://prefix.dev/api/v1/delete/phoenix-dev/linux-64

        with tempfile.TemporaryDirectory() as tmpdir:
            # try to read the subdir repodata.json. If there are no channels in the subdir, the subdir actually
            # doesn't exists and we get an HTTP 404.
            repodata_filepath = Path(tmpdir) / (subdir + "_repodata.json")
            try:
                download_file(
                    args.channel_url + "/" + subdir + "/repodata.json",
                    repodata_filepath,
                    "phoenix-clean-up-schedule",
                    60,
                )
            except HTTPError:
                logger.warning("Could not clean %s/%s subdir, no packages?", args.channel_url, subdir)
                continue

            logger.info("Found repodata.json for %s/%s subdir, cleaning", args.channel_url, subdir)
            failed_all_subdir = False
            # we could read the subdir repodata.json: proceed with the clean-up.
            channel_packages = load_repodata(repodata_filepath)

            # log the packages actually in the channel
            logger.info("Packages on the channel: %s", channel_packages)

            # Filter excluded packages
            current_time = datetime.now(UTC)
            packages_to_delete = select_package_to_delete(
                channel_packages,
                current_time,
                args.min_age_days,
                args.min_number_of_packages,
                args.keep_version_regex,
            )

            # Display the exclude packages
            logger.info("Packages to delete: %s", packages_to_delete)

            clean_up_channel(
                packages_to_delete,  # request fails if there are double /
                args.api_delete_url + ("/" if not args.api_delete_url.endswith("/") else "") + subdir,
                args.token,
                60,
            )

    if failed_all_subdir:
        error_msg = f"Could not clean any sub-directory in channel {args.channel_url}. Is the URL correct?"
        raise RuntimeError(error_msg)

remove_version_matching_regex 

remove_version_matching_regex(package_collection, remove_version_regex)

Remove packages matching a regex from the package collection.

Parameters:

Name Type Description Default
package_collection PackageInfoCollection

Collection of packages to filter

 required

Returns:

Type Description
PackageInfoCollection

Collection of packages with the packages matching the regex removed
Source code in src/phoenixpackagecleanup/clean_channel.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def remove_version_matching_regex(
    package_collection: PackageInfoCollection, remove_version_regex: str
) -> PackageInfoCollection:
    """Remove packages matching a regex from the package collection.

    Parameters
    ----------
    package_collection : PackageInfoCollection
        Collection of packages to filter

    Returns
    -------
    PackageInfoCollection
        Collection of packages with the packages matching the regex removed
    """
    filtered_collection = PackageInfoCollection(defaultdict(list))

    for package_name, packages_info in package_collection.packages_info.items():
        # Filter packages : packages not matching the remove_version_regex are not added to the output collection
        filtered_packages = [pkg for pkg in packages_info if not re.match(remove_version_regex, pkg.version)]
        if filtered_packages:
            filtered_collection.packages_info[package_name] = filtered_packages

    return filtered_collection

select_package_to_delete 

select_package_to_delete(channel_packages, current_time, delete_older_than_days, min_number_of_packages, keep_version_regex)

Select the packages to clean-up from a package collection based on rules inspired by gitlab's clean-up rules.

The selection rules work as the following: - for each package name, get the list of package version/archives - exclude the min_number_of_packages most recent packages from the list - exclude all packages that have been uploaded more recently than delete_older_than_days days old. - what remains in the list is selected for deletion

Parameters:

Name Type Description Default
channel_packages PackageInfoCollection

Collection of packages to clean-up.

 required
current_time datetime

The time at which the script is running, used to remove packages based on upload timestamp

 required
delete_older_than_days int

Packages which upload timestamp is less than delete_older_than_days days old are not considered for deletion

 required
min_number_of_packages int

Only if at least min_number_of_packages are available for a package will the packages be considered for deletion

 required
keep_version_regex str

regex applied to packages versions: if the package version match this, it will NOT be considered for deletion.

 required

Returns:

Type Description
PackageInfoCollection

Collection of packages that should be deleted to clean-up.
Source code in src/phoenixpackagecleanup/clean_channel.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
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
def select_package_to_delete(
    channel_packages: PackageInfoCollection,
    current_time: datetime,
    delete_older_than_days: int,
    min_number_of_packages: int,
    keep_version_regex: str,
) -> PackageInfoCollection:
    """Select the packages to clean-up from a package collection based on rules inspired by gitlab's clean-up rules.

    The selection rules work as the following:
    - for each package name, get the list of package version/archives
    - exclude the `min_number_of_packages` most recent packages from the list
    - exclude all packages that have been uploaded more recently than `delete_older_than_days` days old.
    - what remains in the list is selected for deletion

    Parameters
    ----------
    channel_packages : PackageInfoCollection
        Collection of packages to clean-up.
    current_time : datetime
        The time at which the script is running, used to remove packages based on upload timestamp
    delete_older_than_days : int
        Packages which upload timestamp is less than `delete_older_than_days` days old are not considered for deletion
    min_number_of_packages : int
        Only if at least `min_number_of_packages` are available for a package will the packages be considered for
        deletion
    keep_version_regex : str
        regex applied to packages versions: if the package version match this, it will NOT be considered for
        deletion.

    Returns
    -------
    PackageInfoCollection
        Collection of packages that should be deleted to clean-up.
    """
    # remove packages matching the keep_version_regex
    filtered_by_regex_pkgs = remove_version_matching_regex(channel_packages, keep_version_regex)

    selected_for_deletion_collection = PackageInfoCollection(defaultdict(list))

    for package_name, packages_info in filtered_by_regex_pkgs.packages_info.items():
        # sort package list in decreasing upload time
        packages_sorted_per_timestamp_decreasing = sorted(
            packages_info, key=lambda package_info: package_info.upload_time, reverse=True
        )
        # remove the min_number_of_packages newest package from list, and iterate
        for candidate_for_clean_up in packages_sorted_per_timestamp_decreasing[min_number_of_packages:]:
            if current_time - candidate_for_clean_up.upload_time > timedelta(days=delete_older_than_days):
                selected_for_deletion_collection.packages_info[package_name].append(candidate_for_clean_up)

    return selected_for_deletion_collection