Cách lấy danh sách file từ S3 dựa trên ngày sửa đổi gần nhất bằng Boto3 và AWS Resource

Đề bài – Sử dụng thư viện boto3 trong Python để lấy danh sách các file từ S3, những file được sửa đổi sau một timestamp ngày cho trước.

Ví dụ – Liệt kê test.zip từ Bucket_1/testfolder của S3 nếu nó được sửa đổi sau 2021-01-21 13:19:56.986445+00:00.

Phương pháp/Thuật toán giải quyết

Bước 1 – Import boto3 và các ngoại lệ từ botocore để xử lý lỗi.

Bước 2 – s3_path và last_modified_timestamp là hai tham số trong hàm list_all_objects_based_on_last_modified. "last_modified_timestamp" phải ở định dạng "2021-01-22 13:19:56.986445+00:00". Mặc định boto3 hiểu múi giờ UTC bất kể vị trí địa lý.

Bước 3 – Kiểm tra s3_path có đúng định dạng AWS s3://bucket_name/key không.

Bước 4 – Tạo phiên AWS (session) bằng thư viện boto3.

Bước 5 – Tạo resource AWS cho S3.

Bước 6 – Liệt kê tất cả các đối tượng của prefix cho trước bằng hàm list_objects và xử lý ngoại lệ nếu có.

Bước 7 – Kết quả của hàm trên là một dictionary và chứa tất cả thông tin cấp file trong key tên là 'Contents'. Trích xuất chi tiết cấp bucket vào một đối tượng.

Bước 8 – Đối tượng cũng là một dictionary chứa tất cả chi tiết của file. Lấy chi tiết LastModified của từng file và so sánh với timestamp ngày cho trước.

Bước 9 – Nếu LastModified lớn hơn timestamp cho trước, lưu tên file hoàn chỉnh, ngược lại bỏ qua.

Bước 10 – Trả về danh sách các file được sửa đổi sau timestamp cho trước.

Ví dụ

Mã sau lấy danh sách file từ AWS S3 dựa trên timestamp ngày sửa đổi gần nhất:

import boto3
from botocore.exceptions import ClientError

def list_all_objects_based_on_last_modified(s3_files_path, last_modified_timestamp):
    if 's3://' not in s3_files_path:
        raise Exception('Đường dẫn không phải là đường dẫn S3 hợp lệ.')
    session = boto3.session.Session()
    s3_resource = session.resource('s3')
    bucket_token = s3_files_path.split('/')
    bucket = bucket_token[2]
    folder_path = bucket_token[3:]
    prefix = ""
    for path in folder_path:
        prefix = prefix + path + '/'
    try:
        result = s3_resource.meta.client.list_objects(Bucket=bucket, Prefix=prefix)
    except ClientError as e:
        raise Exception("Lỗi client boto3 trong list_all_objects_based_on_last_modified: " + e.__str__())
    except Exception as e:
        raise Exception("Lỗi không mong đợi trong list_all_objects_based_on_last_modified của s3 helper: " + e.__str__())
    filtered_file_names = []
    for obj in result['Contents']:
        if str(obj["LastModified"]) >= str(last_modified_timestamp):
            full_s3_file = "s3://" + bucket + "/" + obj["Key"]
            filtered_file_names.append(full_s3_file)
    return filtered_file_names

# cung cấp timestamp để lấy test.zip
print(list_all_objects_based_on_last_modified("s3://Bucket_1/testfolder", "2021-01-21 13:19:56.986445+00:00"))
# cung cấp timestamp không file nào được sửa đổi sau đó
print(list_all_objects_based_on_last_modified("s3://Bucket_1/testfolder", "2021-01-21 13:19:56.986445+00:00"))

Kết quả

# cung cấp timestamp để lấy test.zip
[s3://Bucket_1/testfolder/test.zip]
# cung cấp timestamp không file nào được sửa đổi sau đó
[]