如何使用Python删除JFrog Artifactory中超过24小时的文件？

一、使用官方Python SDK（jfrog-client-python） #

如果你的虚拟机允许通过pip安装第三方库，官方的jfrog-client-python是最贴合Python风格的选择——它封装了Artifactory的API调用，无需手动处理HTTP请求细节。

1. 安装依赖：

pip install jfrog-client-python

2. 示例代码：

from jfrog import Artifactory
from datetime import datetime, timedelta

# 初始化Artifactory客户端
artifactory = Artifactory(
    url="https://my.artifactory.instance/artifactory",
    username="your_username",
    password="your_password_or_api_key"
)

# 计算24小时前的毫秒级时间戳（Artifactory默认使用毫秒时间戳）
cutoff_time = int((datetime.now() - timedelta(hours=24)).timestamp() * 1000)

# 构造AQL查询，筛选目标仓库中超过24小时未修改的文件
aql_query = f"""
items.find(
    {{
        "repo": "repo_to_search",
        "type": "file",
        "modified": {{ "$lt": {cutoff_time} }}
    }}
).include("repo", "path", "name")
"""

# 执行AQL查询并遍历结果删除文件
results = artifactory.aql(aql_query)
for item in results["results"]:
    file_path = f"{item['repo']}/{item['path']}/{item['name']}"
    print(f"Deleting {file_path}")
    artifactory.delete(file_path)

二、直接使用requests库（无额外SDK依赖） #

如果无法安装官方SDK，用requests库直接调用Artifactory API是更轻量化的Pythonic方案，比os.system(curl)更易维护、更便于处理错误。

1. 确保requests已安装（若权限允许，执行pip install requests；若无法安装第三方库，可改用标准库urllib，核心逻辑一致）

2. 示例代码：

import requests
from datetime import datetime, timedelta

# Artifactory基础配置
ARTIFACTORY_URL = "https://my.artifactory.instance/artifactory"
TARGET_REPO = "repo_to_search"
AUTH = ("your_username", "your_password_or_api_key")

# 计算24小时前的毫秒级时间戳
cutoff_time = int((datetime.now() - timedelta(hours=24)).timestamp() * 1000)

# 构造AQL查询请求体
aql_payload = {
    "query": f"""
    items.find(
        {{
            "repo": "{TARGET_REPO}",
            "type": "file",
            "modified": {{ "$lt": {cutoff_time} }}
        }}
    ).include("repo", "path", "name")
    """
}

# 发送AQL查询获取待删除文件列表
aql_response = requests.post(
    f"{ARTIFACTORY_URL}/api/search/aql",
    json=aql_payload,
    auth=AUTH
)
aql_response.raise_for_status()  # 主动抛出HTTP请求错误
results = aql_response.json()["results"]

# 遍历删除每个文件
for item in results:
    file_full_url = f"{ARTIFACTORY_URL}/{item['repo']}/{item['path']}/{item['name']}"
    print(f"Deleting {file_full_url}")
    delete_response = requests.delete(file_full_url, auth=AUTH)
    delete_response.raise_for_status()

三、关键注意事项 #

• 认证方式：支持用户名+密码或API密钥（API密钥更安全，可在Artifactory个人设置中生成）
• 分页处理：若符合条件的文件数量极多，AQL默认仅返回100条结果，需添加.limit()和.offset()实现分页查询，避免请求超时
• 错误处理：示例中使用raise_for_status()抛出HTTP错误，实际场景可根据需求添加重试、日志记录等逻辑

内容的提问来源于stack exchange，提问作者Rafa S