Author: Michael Moore is a Certified Master Anaplanner and Director, Enterprise Planning at HP.
In this article I will cover:
- Why use Python for user administration?
- Prepare
- Begin OAuth2 login
- Finish OAuth2 login using redirect URL to fetch authentication token
- Import useful Python modules
- Fetch data about yourself
- Sort model users file by login
- Write a file with all tenant users
- Export user files from selected models
- Find employees who are visitors to this tenant
- Find inactive users to disable
- Find users without selective access
Community Note: there is a downloadable PDF version of this article (at the end), as sometimes the code doesn't present properly in this article.
Why use Python for user administration?
Managing users for a large company with multiple Anaplan models can be a challenge. This article shows how to use Python with the Anaplan API and Anaplan export actions can be used to report on Anaplan users.
The sample reports provided are rough, either writing CSV files or returning a Python list of results. These have been developed as needed over the past year based on problems encountered by the Anaplan COE team at HP. You are welcome to use them directly, or take inspiration from these examples to write your own.
Prepare
Begin OAuth2 login
Login to Anaplan using OAuth2. This code uses the Anaplan SDK Python interface to various Anaplan APIs. See the Anaplan SDK Authentication Guide for options on how to login via Python.
The login local Python module is a simple file that encapsulates reusable OAuth2 login code for multiple notebooks.
import anaplan_sdk
from login import login, redirect
login()
Finish OAuth2 login using redirect url to fetch authentication token
Save the URL from login into value of redirect_response below. OAuth2 is designed to be used for web applications. This step copies the browser URL to a string for use in this notebook.
# redirect_response = 'https://.....?code=...&state=...'
anaplan: anaplan_sdk.Client = redirect(redirect_response)
Import useful Python modules
These libraries are used in the following code examples, mostly for processing CSV files exported from Anaplan.
The local Python module datestamp_file contains code to append today's date to a filename.
import csv
import datetime
from io import StringIO
import concurrent.futures
import logging
import datestamp_file
logger = logging.getLogger(__name__)
Fetch data about yourself
Find out what Anaplan knows about you. There are two API calls to fetch information about an individual user, one in the integration API and one in the SCIM API (System for Cross-domain Identity Management).
me = anaplan.audit.get_user()
scim_me = anaplan.scim.get_user(me.id)
Sort model users file by login
Suppose we want to know which users in a specific model have not logged in lately. This may be part of a process to manage user licenses.
This process starts with exporting a full list of users from a model. User emails are in the first column in this CSV file.
with open("Users.csv", "r", encoding = "utf-8-sig") as f:
reader = csv.reader(f)
headers = next(reader)
emails = [row[0] for row in reader]
We could use fetch data for each user one by one, but it is faster to fetch data for all the users in the tenant and filter that list using the email addresses of users from the model export file.
tenant_users = anaplan.audit.get_users()
users = [u for u in tenant_users if u.email in emails]
Sort users by their last login date. Users who have never logged in have an empty string as their last login date.
def login_key(user):
login = user.last_login_date
if login:
return login #datetime.datetime.fromisoformat(login)
return "" #datetime.datetime.fromtimestamp(0)
inactive = sorted(users, key=login_key)
Write user records ordered by last login date showing their email address and whether their account is active (enabled).
with open("APJ Channel logins H2FY25.csv", "w", encoding="utf-8", newline="") as f:
writer = csv.writer(f)
writer.writerow(["emails", "last_login", "active"])
writer.writerows([user.email, user.last_login_date, user.active]
for user in inactive)
Write a file with all tenant users
Show their email address, last login date, and whether their account is active (enabled).
tenant_users = anaplan.audit.get_users()
with open(datestamp_file.timestamp_filename("TenantUsers"),
"w",
encoding="utf-8",
newline="") as f:
writer = csv.writer(f)
writer.writerow(["emails", "last_login", "active"])
writer.writerows([user.email, user.last_login_date, user.active]
for user in tenant_users)
Export user files from selected models
For all production models, export a user file. This may be helpful for user administration or reporting on user adoption.
Read a current list of production models and export user files from each model (we maintain this data as a saved view in an Anaplan model). The production model records include workspace and model IDs, user export ID, and filename prefix to write.
This implementation uses concurrent.futures to start all the exports in parallel. This could be simpler using the anaplan-sdk.AsyncClient.
ADOPTION_CONFIG = 116000000000 # Adoption Config export action
def write_user_file(filename, export_id, workspace_id, model_id, **params):
"""When user export is finished, write the user file"""
action_id = int(export_id)
data = anaplan.with_model(
workspace_id=workspace_id,
model_id=model_id
).export_and_download(action_id)
if os.path.dirname(filename):
os.makedirs(os.path.dirname(filename), exist_ok=True)
with open(filename, "wb") as f:
f.write(data)
def user_file(config):
"""Run user export for one model"""
config["filename"] = datestamp_file.timestamp_filename(config["rename"])
write_user_file(**config)
def json_row(headers, row):
result = {}
for col, header in enumerate(headers):
result[header] = row[col]
return result
def get_adoption_config():
"""Get production models and user file names from running
an Anaplan model export"""
byte_data = anaplan.export_and_download(ADOPTION_CONFIG)
decoded = byte_data.decode("utf-8")
f = StringIO(decoded)
reader = csv.reader(f, delimiter=",")
headers = next(reader)
config_json = [json_row(headers, row) for row in reader]
return config_json
def all_user_files():
"""Speed this up by running requests concurrently
https://docs.python.org/3/library/concurrent.futures.html#threadpoolexecutor-example
"""
configs = get_adoption_config()
with concurrent.futures.ThreadPoolExecutor() as executor:
# Start the load operations and mark each future with its URL
future_to_config = {
executor.submit(user_file, config): config for config in configs
}
for future in concurrent.futures.as_completed(future_to_config):
config = future_to_config[future]
try:
data = future.result()
except Exception as exc:
logger.error("%r generated an exception: %s" %(config["model"], exc))
else:
logger.debug("%r model done" %(config["model"]))
Run the process to export user files.
all_user_files()
Find employees who are visitors to this tenant
Tenant visitors are listed as users in models, but are not in API results because their default tenant is does not match the model tenant. To find these users, compare the model users to the tenant users.
The mechanism above for exporting user files is patched to collect a list of all these users instead.
model_users = set()
def write_user_file(filename, export_id, workspace_id, model_id, **params):
"""When user export is finished, write the user file"""
action_id = int(export_id)
byte_data = anaplan.with_model(
workspace_id=workspace_id,
model_id=model_id
).export_and_download(action_id)
decoded = byte_data.decode("utf-8")
f = StringIO(decoded)
reader = csv.reader(f, delimiter=",")
headers = next(reader) # ignore headers
for row in reader:
email = row[0]
model_users.add(email)
Fetch the list of customer_visitors to move to our tenant.
CUSTOMER_SUFFIX = "@hp.com"
tenant_users = anaplan.audit.get_users()
tenant_user_set = {user.email for user in tenant_users}
visitor_set = model_users - tenant_user_set
customer_visitors = {email for email in visitor_set if CUSTOMER_SUFFIX in email}
Find inactive users to disable
Idlers are users who have not logged in to Anaplan ever or within the previous six months.
Collect information about all tenant users from Integration API and SCIM API. Remember that visitors to this tenant do not appear in API results.
tenant_users = anaplan.audit.get_users()
scim_users = anaplan.scim.get_users()
tenant_dict = {u.email : u for u in tenant_users}
Test each user to see whether they have been idle long enough to disable. Ignore the following users:
- with no entitlements (no access to any workspaces)
- users who are already disabled
- users with entitlements in workspaces that do not contain the workspace code
- users with accounts created within
CREATE_THRESHOLD (30 days)
- users with last login within
LOGIN_THRESHOLD (180 days)
CREATE_THRESHOLD = datetime.timedelta(days=30)
LOGIN_THRESHOLD = datetime.timedelta(days=180)
NOW = datetime.datetime.now(datetime.timezone.utc)
def idler(scim_user, workspace_code: str = "") -> bool:
"""Find candidate users to disable.
Arguments:
scim_user -- user record from SCIM API
workspace_code -- filter workspace names that contain this string
"""
if not scim_user.entitlements:
return False
if not scim_user.active:
return False
entitlements = scim_user.entitlements[-1].value.split(',')
if any(workspace_code not in name for name in entitlements):
return False
created = datetime.datetime.fromisoformat(scim_user.meta.created)
audit_user = tenant_dict[scim_user.user_name]
if not audit_user.last_login_date and NOW - created < CREATE_THRESHOLD:
return False
if not audit_user.last_login_date:
return True
login = datetime.datetime.fromisoformat(audit_user.last_login_date)
if NOW - login < LOGIN_THRESHOLD:
return False
return True
Write a file with all the users who meet the idle criteria above.
def write_idle_users_file(workspace_code: str = ""):
"""Write CSV file with idle user records.
Arguments:
workspace_code -- filter workspace names that contain this string
"""
idle_users = [u for u in scim_users if idler(u, workspace_code)]
with open(datestamp_file.timestamp_filename(f"{workspace_code}_IdleUsers"),
"w",
encoding="utf-8",
newline="") as f:
writer = csv.writer(f)
f.write(
f"Users in {workspace_code}"
" created more than {CREATE_THRESHOLD.days} days ago"
" and idle for more than {LOGIN_THRESHOLD.days} days"
)
writer.writerow([])
writer.writerow(["emails", "created", "last_login", "active"])
writer.writerows(
[
tenant_dict[user.user_name].email,
user.meta.created,
tenant_dict[user.user_name].last_login_date,
tenant_dict[user.user_name].active
]
for user in idle_users
)
Find users without selective access
In models with selective access, it is not typical for users to not have settings for any of the dimensions. Identify those users.
Each production model has a user export action named "Grid - Users.csv". Run that action and filter user records by those who have no selective access settings.
EXCLUDE_ROLES = ['No Access', 'Full Access']
USER_EXPORT_NAME = "Grid - Users.csv"
def read_user_export(workspace_id, model_id) -> list:
"""Create list of user dict from user export"""
client = anaplan.with_model(
workspace_id=workspace_id,
model_id=model_id
)
exports = client.get_exports ()
user_export = [e for e in exports if e.name == USER_EXPORT_NAME]
if not user_export:
logger.error ("ERROR: user export not found")
return []
action_id = int(user_export[0].id)
byte_data = client.export_and_download(action_id)
decoded = byte_data.decode("utf-8")
f = StringIO(decoded)
reader = csv.reader(f, delimiter=",")
headers = next(reader)
users = [json_row(headers, row) for row in reader]
return users
def missing_user (user):
if user['Model Role'] in EXCLUDE_ROLES:
return False
return not any(user[k] for k in selective_access)
def users_missing_selective_access (workspace_id, model_id):
model_users = read_user_export (workspace_id, model_id)
selective_access = list (model_users[0].keys ())[4:-3] if model_users else []
missing_sa_users = [u for u in model_users if missing_user (u)]
return missing_sa_users
Fetch the list of missing_sa_users without selective access settings.
model_id = "28888E705A6F4EDBB40AAC14CCFBABAD" # Quota Deployment CEE [PROD - H1 FY26]
workspace_id = "8a868cd88e886882018fef99a2240112" # HPI Quota Deployment PROD
missing_sa_users = users_missing_selective_access(workspace_id, model_id)
PDF
Questions? Leave a comment!
