Get an email when someone joins your Slack workspace

from clickruntime import on_trigger, credentials, http, state, notify

WORKSPACE = "Acme"

def slack(method, **params):
    resp = http.get(f"https://slack.com/api/{method}", params=params,
                    headers={"Authorization": f"Bearer {credentials.connection('slack', 'work')}"})
    data = resp.json()
    if not data.get("ok"):
        raise RuntimeError(data.get("error"))
    return data

def account_type(u):
    if u.get("is_ultra_restricted"):
        return "single-channel guest"
    if u.get("is_restricted"):
        return "multi-channel guest"
    return "admin" if u.get("is_admin") else "member"

@on_trigger
def run(event):
    users, cursor = [], None
    while True:
        page = slack("users.list", limit=200, **({"cursor": cursor} if cursor else {}))
        users += [u for u in page["members"] if not u.get("is_bot") and not u.get("deleted") and u["id"] != "USLACKBOT"]
        cursor = page.get("response_metadata", {}).get("next_cursor")
        if not cursor:
            break
    prev = state.get("ids")
    state.set("ids", [u["id"] for u in users])
    if prev is None:
        return {"baseline": len(users)}
    new = [u for u in users if u["id"] not in set(prev)]
    if not new:
        return {"new": 0}
    lines = []
    for u in new:
        p = u.get("profile", {})
        lines.append(" · ".join(x for x in [p.get("real_name") or u["name"], p.get("email"), p.get("title"),
                                             u.get("tz"), account_type(u)] if x))
    notify.email_me(f"{len(new)} new {'person' if len(new) == 1 else 'people'} in the {WORKSPACE} Slack", "\n".join(lines))
    return {"new": len(new)}