User Reporting

Introduction

There are a few metrics gathered for user accounts. These can be presented in a summary form or in a detailed form. The information collected about users includes if the account is active (enabled) or not, what roles the user has been granted, and what data sets the users have access to. This data is summarized daily.

UserSummary

Get a summary of users and roles.

Inputs

date (datetime): The specific day to summarize

Outputs

users (UserSummarycounts): Counts of enabled/disabled users
app_roles (List[AppRoles]): List of all available user roles

Try It Out

You can try outGetUserSummary below.

from pprint import pprint
from indico import IndicoClient, IndicoConfig

from indico.queries.usermetrics import GetUserSummary
from indico.types.user_metrics import UserSummary

HOST = "my-cluster.indico.io"
TOKEN = "./indico_api_token.txt"
my_config = IndicoConfig(host=HOST, api_token_path=TOKEN)
client = IndicoClient(config=my_config)

summary: UserSummary = client.call(GetUserSummary())
  
roles = {role.role: role.count for role in summary.app_roles}
users = {"enabled": summary.users.enabled, "disabled": summary.users.disabled}
pprint(
    {
        "roles": roles,
        "users": users,
    }
)

User Snapshots

Gets the status of a user's roles and permissions on a specific date.

from pprint import pprint
from indico import IndicoClient, IndicoConfig

from indico.queries.usermetrics import GetUserSummary
from indico.types.user_metrics import UserSummary

HOST = "my-cluster.indico.io"
TOKEN = "./indico_api_token.txt"
my_config = IndicoConfig(host=HOST, api_token_path=TOKEN)
client = IndicoClient(config=my_config)

summary: UserSummary = client.call(GetUserSummary())
  
roles = {role.role: role.count for role in summary.app_roles}
users = {"enabled": summary.users.enabled, "disabled": summary.users.disabled}
pprint(
    {
        "roles": roles,
        "users": users,
    }
)

User Changelog

This generates a log of events over time representing changes in user permissions at an app and dataset level. Users can paginate over the results of this call.

Inputs

filters (UserSnapshotFilter): Filter the query based on UserMetricsFilter criteria
start_date (datetime): Specific start date for query
end_date (datetime): Specific end date for query
limit (int): Limit how many come back per query or per page

Outputs

ID (str): ID of the log entry
date (datetime): Time of long entry
user_id (int): ID of the user whose attributes were changed
user_email (str): Email of the user whose attributes were changed
updated_by (ID): ID of the account which made the change
updater_email (str): Email of the account which made the change
previously_enabled (bool): True if this account was enabled prior to change
enabled (bool): True if account is enabled after the change
previous_roles (List[str]): Roles assigned prior to the change
roles (List[str]): Roles assigned after the change
previous_datasets (List[DatasetRole]): List of dataset/role mappings prior to the change
datasets (List[DatasetRole]): List of dataset/role mappings after the change
changes_made (List[str]): A list of changed made

Try It Out

from indico import IndicoClient, IndicoConfig
from indico.queries.usermetrics import GetUserChangelog, UserMetricsFilter
from datetime import datetime
from indico.types.user_metrics import UserChangelog

HOST = "my-cluster.indico.io"
TOKEN = "./indico_api_token.txt"
my_config = IndicoConfig(host=HOST, api_token_path=TOKEN)
client = IndicoClient(config=my_config)

# filters
USER_ID = 26
USER_EMAIL = "[email protected]"
filter_id = UserMetricsFilter(user_id=USER_ID)
filter_email = UserMetricsFilter(user_email=USER_EMAIL)
start_date = datetime.strptime("1-1-2023", "%m-%d-%Y")
end_date = datetime.now()
limit = 100

# for date range
log: UserChangelog
for page in client.paginate(GetUserChangelog(start_date=start_date, end_date=end_date)):
    for log in page:
        pass

# for daterange with user filter
log: UserChangelog
for page in client.paginate(
    GetUserChangelog(start_date=start_date, end_date=end_date, filters=filter_id)
):
    for log in page:
        pass

# for daterange with user filter
# limit by page size
log: UserChangelog
for page in client.paginate(
    GetUserChangelog(
        start_date=start_date, end_date=end_date, filters=filter_id, limit=limit
    )
):
    assert len(page) <= limit
    for log in page:
        pass

# total limit
logs = client.call(
    GetUserChangelog(
        start_date=start_date, end_date=end_date, filters=filter_id, limit=limit
    )
)
assert len(logs) <= limit

Generate User Changelog Report

This creates an async job which will generate a file that can be downloaded later. The contents are the same as the prior call.

from indico import IndicoClient, IndicoConfig
from json import dump
from indico.queries import RetrieveStorageObject, JobStatus
from indico.queries.usermetrics import GenerateChangelogReport, UserMetricsFilter
from datetime import datetime, timedelta
from indico.types.user_metrics import UserChangelog
from indico.types.jobs import Job

HOST = "mycluster.indico.io"
TOKEN = "./indico_api_token.txt"
my_config = IndicoConfig(host=HOST, api_token_path=TOKEN)
client = IndicoClient(config=my_config)

# filters
USER_ID = 26
USER_EMAIL = "[email protected]"
filter_id = UserMetricsFilter(user_id=USER_ID)
filter_email = UserMetricsFilter(user_email=USER_EMAIL)
end_date = datetime.now()
start_date = end_date - timedelta(days=7)

format_json = "json"

# csv in date range
changelogs = client.call(
    GenerateChangelogReport(
        start_date=start_date, end_date=end_date, report_format="csv"
    )
)
job: Job = client.call(JobStatus(id=changelogs.job_id, wait=True))
assert job.status == "SUCCESS"
result = client.call(RetrieveStorageObject(job.result))
with open("results.csv", mode="w") as f:
    # save output
    f.write(result)


# with optional user filter
changelogs = client.call(
    GenerateChangelogReport(
        start_date=start_date, end_date=end_date, filters=filter_email
    )
)
job: Job = client.call(JobStatus(id=changelogs.job_id, wait=True))
assert job.status == "SUCCESS"
result = client.call(RetrieveStorageObject(job.result))

with open(f"results_USER_EMAIL.csv", mode="w") as f:
    # save output
    f.write(result)

# format output as json
changelogs = client.call(
    GenerateChangelogReport(
        start_date=start_date, end_date=end_date, report_format=format_json
    )
)
job: Job = client.call(JobStatus(id=changelogs.job_id, wait=True))
assert job.status == "SUCCESS"
result = client.call(RetrieveStorageObject(job.result))

with open("results.json", mode="w") as f:
    # save output
    dump(result, f)