Back up your Airtable base to S3 every night

import csv, io, boto3
from datetime import date, timedelta
from clickruntime import on_trigger, credentials, http, notify

BASE_ID = "appXXXXXXXXXXXXXX"
REGION, BUCKET = "us-east-1", "northwind-files"

def aws(service):
    return boto3.client(service, region_name=REGION, aws_access_key_id=credentials.secret("AWS_ACCESS_KEY_ID"),
                        aws_secret_access_key=credentials.secret("AWS_SECRET_ACCESS_KEY"))

@on_trigger
def run(event):
    at = {"Authorization": "Bearer " + credentials.secret("AIRTABLE_TOKEN")}
    s3, today, total = aws("s3"), date.today(), 0
    tables = http.get(f"https://api.airtable.com/v0/meta/bases/{BASE_ID}/tables", headers=at).json()["tables"]
    for t in tables:
        records, offset = [], None
        while True:
            page = http.get(f"https://api.airtable.com/v0/{BASE_ID}/{t['id']}", headers=at, params={"offset": offset} if offset else {}).json()
            records += page["records"]
            if not (offset := page.get("offset")):
                break
        out, fields = io.StringIO(), sorted({k for r in records for k in r["fields"]})
        w = csv.DictWriter(out, fieldnames=["id"] + fields)
        w.writeheader()
        w.writerows({"id": r["id"], **{k: str(v) for k, v in r["fields"].items()}} for r in records)
        s3.put_object(Bucket=BUCKET, Key=f"backups/{today}/{t['name']}.csv", Body=out.getvalue().encode())
        total += len(records)
    old = [o for o in s3.list_objects_v2(Bucket=BUCKET, Prefix="backups/").get("Contents", []) if o["Key"].split("/")[1] < str(today - timedelta(days=30))]
    if old:
        s3.delete_objects(Bucket=BUCKET, Delete={"Objects": [{"Key": o["Key"]} for o in old]})
    notify.slack(f":white_check_mark: Backup done: {len(tables)} tables, {total} records to s3://{BUCKET}/backups/{today}/ ({len(old)} old files removed)", channel="#ops")
    return {"tables": len(tables), "records": total}