Publish the next ready video from your Airtable queue

import io
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from googleapiclient.http import MediaIoBaseUpload
from clickruntime import on_trigger, http, credentials, notify

def api(name, version):
    return build(name, version, credentials=Credentials(credentials.connection("google", "work")), cache_discovery=False)

AT = "https://api.airtable.com/v0/appXXXXXXXXXXXXXX"

def at(method, path, **kw):
    r = http.request(method, f"{AT}/{path}", headers={"Authorization": f"Bearer {credentials.connection('airtable', 'work')}"}, **kw)
    return r.raise_for_status() or r.json()

@on_trigger
def run(event):
    records = at("GET", "Content", params={"filterByFormula": "{Status}='Ready'", "maxRecords": 1, "sort[0][field]": "Publish date"})["records"]
    if not records:
        return {"published": False}
    rec = records[0]
    f = rec["fields"]
    media = MediaIoBaseUpload(io.BytesIO(http.get(f["Video"][0]["url"], timeout=120).content), mimetype="video/mp4", resumable=True)
    video = api("youtube", "v3").videos().insert(part="snippet,status", media_body=media, body={
        "snippet": {"title": f["Title"], "description": f.get("Description", "")}, "status": {"privacyStatus": "public"}}).execute()
    link = f"https://youtu.be/{video['id']}"
    at("PATCH", "Content", json={"records": [{"id": rec["id"], "fields": {"Status": "Published", "YouTube link": link}}]})
    notify.email_me(f"Published: {f['Title']}", link)
    return {"video": video["id"]}