User Reporting
Basic statistical information about user accounts.
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
Input
GetUserSummary Inputs
date (datetime): specific day to summarize.
Output
GetUserSummary 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
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
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)
Updated 7 months ago