Get Users by Role

users_by_role.py

from gql import Client, gql
from gql.transport.requests import RequestsHTTPTransport
import os
import pandas as pd
from datetime import datetime, timedelta


USER_GRANTS_QUERY = """
query UsersByRole {
     usersOrError {
    	__typename 
      ... on DagsterCloudUsersWithScopedPermissionGrants {
        users {
          id
          user{
            email
          }
          deploymentPermissionGrants {
            grant
          }
        }
      }
    }
}
"""


def run_query(query, ENDPOINT, AUTH_TOKEN):
    """Run a GQL query against a prod cloud instance, see https://your_org.dagster.cloud/prod/graphql"""
    transport = RequestsHTTPTransport(
        url=ENDPOINT, headers={"Dagster-Cloud-Api-Token": AUTH_TOKEN}
    )

    with Client(
        transport=transport,
        fetch_schema_from_transport=True,
    ) as session:
        gql_query = gql(query)
        result = session.execute(gql_query)

    return result


_map_role_level = {"VIEWER": 0, "LAUNCHER": 1, "EDITOR": 2, "ADMIN": 3}

_map_level_role = {v: k for k, v in _map_role_level.items()}


def _get_max_role_by_user(raw_gql_user_grants):
    """Returns a mapping of user email to their max role across deployments"""
    users = raw_gql_user_grants["usersOrError"]["users"]
    users_df = pd.DataFrame.from_dict(users)
    user_roles = []
    user_email = []
    for u in range(len(users_df)):
        user_grants = pd.DataFrame.from_dict(
            users_df.iloc[u]["deploymentPermissionGrants"]
        )
        user_grants["level"] = user_grants["grant"].map(_map_role_level)
        max_user_grant = max(user_grants["level"])
        user_roles.append(_map_level_role[max_user_grant])
        user_email.append(users_df.iloc[u]["user"]["email"])

    users_max_role = pd.DataFrame({"user_email": user_email, "max_role": user_roles})

    return users_max_role


if __name__ == "__main__":
  
    # add these inputs for your org
    ENDPOINT = os.getenv(
        "DAGSTER_CLOUD_GRAPHQL_URL"
    )  # eg https://your_org.dagster.cloud/prod/graphql
    AUTH_TOKEN = os.getenv("DAGSTER_CLOUD_TOKEN")
    
    results = run_query(USER_GRANTS_QUERY, ENDPOINT, AUTH_TOKEN)
    users_max_role = _get_max_role_by_user(results)

    summary = users_max_role.groupby(["max_role"]).count()

    print(
        f""" SUMMARY:
            {summary}
        """
    )

    print(
        f""" DETAILS:
            {users_max_role}
        """
    )