Delete old objects from a bucket using a fake S3 client:
s3.list_objects_v2(Bucket=..., ContinuationToken=...) returns {"Contents": [{"Key", "AgeDays"}], "IsTruncated": bool, "NextContinuationToken": str}.s3.delete_objects(Bucket=..., Delete={"Objects": [{"Key": k}, ...]}) deletes up to batch_size
keys per call (S3's real limit is 1000).Delete every object older than max_age_days whose key does not start with any prefix in keep.
Return the total number deleted. The harness checks your batches.
Implement cleanup. Input: s3 (fake), bucket: str, max_age_days: int, keep: list[str], batch_size: int. Returns: int.
This code runs before yours (fake clients, clocks and the test harness). You can use its names; don't edit them.
class FakeS3:
def __init__(self, pages, batch_limit):
self.pages, self.batch_limit, self.deleted, self.batches = pages, batch_limit, [], []
def list_objects_v2(self, Bucket, ContinuationToken=None):
i = 0 if ContinuationToken is None else int(ContinuationToken)
page = {"Contents": self.pages[i], "IsTruncated": i + 1 < len(self.pages)}
if page["IsTruncated"]:
page["NextContinuationToken"] = str(i + 1)
return page
def delete_objects(self, Bucket, Delete):
objs = Delete["Objects"]
if not objs or len(objs) > self.batch_limit:
raise ValueError(f"MalformedXML: {len(objs)} objects in one request")
self.batches.append(len(objs))
self.deleted += [o["Key"] for o in objs]
return {"Deleted": objs}
def __harness__(fn, pages, max_age, keep, batch):
s3 = FakeS3(pages, batch)
n = fn(s3, "logs", max_age, keep, batch)
return {"returned": n, "deleted": sorted(s3.deleted), "batches": s3.batches}
def cleanup(s3, bucket: str, max_age_days: int, keep: list[str], batch_size: int) -> int:
return 0
Run executes the public tests. Submit grades against public and hidden tests.
Delete old objects from a bucket using a fake S3 client:
s3.list_objects_v2(Bucket=..., ContinuationToken=...) returns {"Contents": [{"Key", "AgeDays"}], "IsTruncated": bool, "NextContinuationToken": str}.s3.delete_objects(Bucket=..., Delete={"Objects": [{"Key": k}, ...]}) deletes up to batch_size
keys per call (S3's real limit is 1000).Delete every object older than max_age_days whose key does not start with any prefix in keep.
Return the total number deleted. The harness checks your batches.
Implement cleanup. Input: s3 (fake), bucket: str, max_age_days: int, keep: list[str], batch_size: int. Returns: int.
This code runs before yours (fake clients, clocks and the test harness). You can use its names; don't edit them.
class FakeS3:
def __init__(self, pages, batch_limit):
self.pages, self.batch_limit, self.deleted, self.batches = pages, batch_limit, [], []
def list_objects_v2(self, Bucket, ContinuationToken=None):
i = 0 if ContinuationToken is None else int(ContinuationToken)
page = {"Contents": self.pages[i], "IsTruncated": i + 1 < len(self.pages)}
if page["IsTruncated"]:
page["NextContinuationToken"] = str(i + 1)
return page
def delete_objects(self, Bucket, Delete):
objs = Delete["Objects"]
if not objs or len(objs) > self.batch_limit:
raise ValueError(f"MalformedXML: {len(objs)} objects in one request")
self.batches.append(len(objs))
self.deleted += [o["Key"] for o in objs]
return {"Deleted": objs}
def __harness__(fn, pages, max_age, keep, batch):
s3 = FakeS3(pages, batch)
n = fn(s3, "logs", max_age, keep, batch)
return {"returned": n, "deleted": sorted(s3.deleted), "batches": s3.batches}
def cleanup(s3, bucket: str, max_age_days: int, keep: list[str], batch_size: int) -> int:
return 0
Run executes the public tests. Submit grades against public and hidden tests.