Remove Users from Dagster+ organization

remove_dagster_users.py

#!/usr/bin/env python3
#!/usr/bin/env python3
# /// script
# dependencies = [
#   "requests<3",
# ]
# ///
import argparse
import sys
import requests

REMOVE_USER_MUTATION = (
    "mutation RemoveUserFromOrganization($email: String!) {"
    "  removeUserFromOrganization(email: $email) {"
    "    __typename"
    "  }"
    "}"
)

def graphql_endpoint_for_org(org: str) -> str:
    return f"https://{org}.dagster.cloud/graphql"


def remove_user(session: requests.Session, endpoint: str, email: str) -> tuple[bool, str]:
    payload = {
        "operationName": "RemoveUserFromOrganization",
        "query": REMOVE_USER_MUTATION,
        "variables": {"email": email},
    }

    r = session.post(endpoint, json=payload, timeout=30)
    r.raise_for_status()
    data = r.json()

    if "errors" in data and data["errors"]:
        return False, data["errors"][0].get("message", str(data["errors"]))

    typename = data["data"]["removeUserFromOrganization"]["__typename"]
    if "Success" in typename:
        return True, f"Removed {email}"
    return False, f"Not removed {email} ({typename})"


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Remove users from a Dagster Cloud organization"
    )
    parser.add_argument(
        "--org",
        required=True,
        help="Dagster Cloud organization slug (e.g. my-company)",
    )
    parser.add_argument(
        "--token",
        required=True,
        help="Dagster Cloud user or API token with org admin permissions",
    )
    parser.add_argument(
        "emails",
        nargs="+",
        help="Email addresses of users to remove",
    )

    args = parser.parse_args()
    endpoint = graphql_endpoint_for_org(args.org)

    session = requests.Session()
    session.headers.update(
        {
            "Authorization": f"Bearer {args.token}",
            "Content-Type": "application/json",
        }
    )

    all_ok = True
    for email in args.emails:
        ok, msg = remove_user(session, endpoint, email)
        print(("✅ " if ok else "❌ ") + msg, file=sys.stdout if ok else sys.stderr)
        all_ok = all_ok and ok

    sys.exit(0 if all_ok else 1)


if __name__ == "__main__":
    main()

remove_users_grapql.graphql

mutation RemoveUserFromOrganization($email: String!) {
  removeUserFromOrganization(email: $email) {
    __typename
    ... on RemoveUserFromOrganizationSuccess {
      email
    }
    ... on PythonError {
      message
      stack
    }
  }
}

variables

{
  "email": "user@email.com"
}