utils.leanix_client_module
Using this module, you can interact with LeanIX without thinking about the LeanIX API.
Only minimal configuration is required: An API token is already enough, the rest is auto-detected.
Key Features
- It auto-detects your LeanIX host name and workspace
- It auto-detects your data model, i.e. all fact sheet types and their relations, which is very useful especially if you are running a non-standard data model
- It can boost your script performance by preloading all fact sheets of all types (or only selected types) into memory on startup, so that your script can operate on local data without fetching any data from LeanIX during runtime.
Examples
# Load all Application fact sheets
> applications = Application.all()
# Get their displayNames
> [app["displayName"] for app in applications]
['ERP - Global', 'HR - Global Human Resources System', 'LeanIX', 'Legacy ERP']
# Get subscriptions for a specific role
> app["subscriptions"].get("System Owner", []) # returns [] if fact sheet has no subscriptions in this role
[{'email': 'john.doe@example.com', 'id': '028b98f1-36da...', 'type': 'RESPONSIBLE', 'comment': 'test'},
{'email': 'jane.doe@example.com', 'id': '36da028b-98f1...', 'type': 'RESPONSIBLE', 'comment': 'test2'}]
# Get fact sheet by name or ID
> user_group = UserGroup("Acme Holding SE")
> user_group = UserGroup("3263428a-5841-409e-a8e2-22c334b4da53")
# Follow relations of a fact sheet
> app1.businesscapabilities()
[Controlling, Finance, Logistics, Materials Management, Sales] # List of BusinessCapability fact sheets
# Get the displayName of all processes related to an application
> [p["displayName"] for p in Application("ERP - Global ECC").processes()]
['Quote-to-Cash', 'Order-to-Invoice']
# Follow long chains of relations
> user_group.applications()[0].itcomponents()[0].providers() ...
# Tag helpers, attribute helpers
> user_group["location"]["geoCountry"]
> if "Holding" in user_group["tags"]["Company"] ...
Data Model
Your LeanIX data model is automatically translated into a Python data model. You can find the Python data model in
the file utils/ws_[workspacename]/fact_sheet_types.py
.
If you made any changes to your LeanIX data model, you can force a re-generation of your Python data model by deleting the Python data model file.
Event Handlers
Besides leanix_client
, we have built a component that makes it easy to define own event handlers that can react to
Webhook events. Have a look at the following event handlers which we have built using our own component:
create_subscription_handler.CreateSubscriptionHandler
to automatically subscribe somebody when they create a new fact sheetfact_sheet_deletion_mail_handler.FactSheetDeletionMailHandler
to receive an e-mail notification when somebody archives a fact sheetsubscription_deletion_mail_handler.SubscriptionDeletionMailHandler
to receive an e-mail notification when somebody removes their subscription from a fact sheet
Quality Reminders
As a key pillar of active data quality management, you need to remind your users whenever their fact sheets
have quality gaps. The quality_issue_reminder.QualityIssueReminder
does exactly that: It applies Quality Checkers
to all users' fact sheets and sends a pretty e-mail summary to those users who need to act.
Quality Reports
To understand your data quality in the first place, a quality reports helps to see which parts of the organization are
still struggling to keep their data quality at an appropriate level. A Quality Report applies all relevant
Quality Checkers
to FactSheet
s and organization.Organization
s, and aggregates their result in an output format.
Reference implementation: data_completeness_per_organization.DataCompletenessPerOrganization
Quality Checkers
A quality checker can be applied to a FactSheet
or an organization.Organization
to evaluate whether those elements
have quality gaps (and if so, which gaps they have).
application_quality_checker.ApplicationQualityChecker
organization_quality_checker.OrganizationQualityChecker
user_quality_checker.UserQualityChecker
Open-Source Licenses
LeanIX Automation Platform depends on open-source software. The open-source license files and the corresponding software package names are listed here: https://download.aronis.de/licenses/
Managing the connection to LeanIX and running all API calls.
Initializes a connection to the workspace provided. You can maintain API tokens for different workspaces in
leanix_config_local.py
, and use workspace_name
(such as AcmeProduction) to identify the one you want to use.
In the automation module, the default_tag_sets
are the standard events that the automation webhook listens to.
Standard:
default_tag_sets = [
["pathfinder", "FACT_SHEET_CREATED"],
["pathfinder", "FACT_SHEET_UPDATED"],
["pathfinder", "FACT_SHEET_ARCHIVED"],
["pathfinder", "FACT_SHEET_DELETED"],
["pathfinder", "SUBSCRIPTION_DELETED"],
["pathfinder", "SUBSCRIPTION_UPDATED"],
["pathfinder", "SUBSCRIPTION_CREATED"],
["pathfinder", "FACT_SHEET_FIELD_UPDATED"],
["pathfinder", "FACT_SHEET_TAG_ADDED"],
["pathfinder", "FACT_SHEET_TAG_REMOVED"],
["pathfinder", "FACT_SHEET_ACCESS_CONTROL_LIST_UPDATED"],
["pathfinder", "RELATION_CREATED"],
["pathfinder", "RELATION_UPDATED"],
["pathfinder", "RELATION_SWITCH"],
["pathfinder", "RELATION_ARCHIVED"],
["pathfinder", "RELATION_DELETED"],
["pathfinder", "DOCUMENT_CREATED"],
]
In edit_webhook_interactively
, you can activate and deactivate event types interactively for each webhook.
As available tag sets, it shows the union of default_tag_sets
and further_tag_sets
.
Only used internally by FactSheet.field()
.
Loads the language model for language language
into LeanIXClient().language_model[language]
.
If users are not yet cached or if refresh is forced by setting force_refresh
:
Loads all available users from LeanIX API and stores them in self.users
as list of User
instances.
The following attributes are loaded: firstName, lastName, displayName, email, status, currentLogin.
Returns the workspace URL, which is used to generate the links in the automatic e-mails.
Auto-detects some workspace properties so that you don't need to configure them:
- user id (of the user who created the token)
- workspace id
Also, it checks if the name of the workspace matches the name in your configuration, just to be sure your API token was configured for the correct workspace.
Generic method for calling any LeanIX API method. Takes care of:
- retry on HTTP error
- reconnecting if the session token has expired
- following a GraphQL cursor when calling the allFactSheets function
- following paginated REST APIs
- recognizing different kinds of errors (business logic or technical) and printing meaningful debug info
path
is the path of the API method, such as '/services/pathfinder/v1/graphql'.
method
is the http method: GET / POST / PUT / DELETE
body
is the request body, either as JSON string or as dict (which will be auto-converted to JSON).
params
is a dict with additional API parameters, such as {"page": 1}
. Transferred as part of the request.
Returns the ["data"]
part of the JSON response.
Generic method to call GraphQL functions.
Turns the JSON query
into a string and calls self.request() to the GraphQL endpoint with HTTP method POST.
Returns the unprocessed response of self.request()
.
(Re-)loads all available subscription roles into self.subscription_roles
.
Loads the user details of the API Token's user into self.user_name_[first|last|display].
.
Warning: This creates a subscription role in the workspace configuration. If you want to subscribe a user to a fact sheet, use FactSheet.add_subscription().
Returns the tag id for a specific tag name and tag group name.
Warning: This deletes a tag from the workspace configuration. If you want to delete a tag from a fact sheet, use FactSheet.delete_tags().
Warning: This deletes a tag from the workspace configuration. If you want to delete a tag from a fact sheet, use FactSheet.delete_tags().
Creates a new tag group with the given properties.
Creates a new tag with the given properties.
Returns all Webhooks subscriptions.
Note: If the Managing User of a Webhook is set to "Current User (private Webhook)", other users cannot see it.
Returns the Webhook subscription with the given ID.
Note: If the Managing User of a Webhook is set to "Current User (private Webhook)", other users cannot see it.
Advances a Pull Webhook's cursor after an event has been processed successfully.
Returns all Webhook deliveries for Webhook webhook_id
. The response can be pretty large, so size=10 is best.
Example:
`LeanIXClient().get_webhook_logs(id, {"sort": "createdAt-asc", "size": 15, "page": 1}, {"page_limit": 1})`
Sometimes LeanIX breaks webhooks. This function fixes a broken webhook by deleting and recreating it. Only works for PUSH webhooks at the moment.
Creates a webhook using a webhook dict structure containing all parameters. Used internally to (re-)create
webhooks. To create webhooks using a more intuitive method, please refer to LeanIXClient.register_webhook()
.
If you edit a webhook manually in the LeanIX UI, it forgets all non-standard event types it was subscribed to.
This function re-adds all non-standard event types (provided by default_tag_sets
) to subscription id
, or,
in case id
is omitted, fixes all webhooks that contain "LeanIX Automation Platform" in their identifier.
Registers a new Webhook.
name
: Name of the new Webhook.
target
: Target URL including protocol and port number.
user
: HTTP user, in case the Webhook uses HTTP basic auth. Set to "None" to switch off HTTP auth.
pwd
: HTTP password, in case the Webhook uses HTTP basic auth.
delivery_type
: "PUSH" or "PULL"
Internally calls LeanIXClient.create_webhook(wh)
.
Updates a Webhook.
webhook
is the dict structure containing the webhook data, since it needs to be passed during the update.
Adds a single new event type to the webhook with the given id
.
Examples:
LeanIXClient().add_webhook_event_type(id, "SUBSCRIPTION_DELETED")
LeanIXClient().add_webhook_event_type(id, "SUBSCRIPTION_CREATED")
LeanIXClient().add_webhook_event_type(id, "FACT_SHEET_FIELD_UPDATED")
Shows an interactive console to easily edit webhooks and add more (unsupported) event types to them.
As available tag sets, it shows the union of default_tag_sets
and further_tag_sets
.
Best run from console.py
or from test_factSheet.py
.
Example:
LeanIXClient().edit_webhook_interactively()
Loads the LeanIX feature configuration for this workspace into LeanIXClient().features.
Returns the configuration of a specific feature for this workspace. See get_features()
for full list.
Creates a metrics schema.
Example:
schema_data = {
"name": "AronisAutomation",
"description": "Automated Test.",
"attributes": [
{"name": "factSheetId", "type": "dimension"}, # Include this to connect the schema to a fact sheet when creating a diagram
{"name": "Description", "type": "metric"},
{"name": "Owning Org", "type": "metric"},
{"name": "App Manager", "type": "metric"},
{"name": "Lifecycle", "type": "metric"},
]
}
LeanIXClient().create_schema(schema=schema_data)
Loads a series of points for the given schema id
.
The optional parameter options["begin"]
and options["end"]
specifies the time range to load.
Note that timestamp needs to be an int, localized to the Europe/London timezone.
Example:
day = datetime.datetime(2024, 1, 1)
timezone = pytz.timezone('Europe/London')
day_localized = timezone.localize(day)
timestamp = int(day_localized.timestamp())
LeanIXClient().get_metric_points(schema_id, {"begin": timestamp - 3600, "end": timestamp})
Gets one single metric point by schema id
and timestamp
. See get_metric_points
for timestamp requirements.
Posts a metric point into the schema with the given id
.
Example:
point = {
"timestamp": timestamp,
"factSheetId": "1234-5678-....",
"Description": randint(0, 100),
"Owning Org": randint(0, 100),
"App Manager": randint(0, 100),
"Lifecycle": randint(0, 100),
}
LeanIXClient().post_metric_point(schema_id, point)
Calls post_metric_point
for every single point in the points
list.
Deletes the metric point with the given timestamp
from the schema with the given id
.
See get_metric_points
for timestamp requirements.
Deletes all metric points within a range.
Currently not supported properly by LeanIX - returns errors (2023-01-15).
Saves the workspace settings as file. This is required before modifying the workspace settings via API call, so
that you can restore the settings in case of an error.
The file is stored as utils/settings.ws_[workspace name].[timestamp].json
.
WARNING - This can break your workspace configuration! Only handle this with care, and always keep a backup
of your workspace settings in get_workspace_settings
handy, so that you can reset them using this method.
Creates a single metrics chart. Currently only used internally for the Data Quality Management report.
Creates all KPI charts for the Data Quality Management report. Only used internally at the moment.
Returns all technical users.
Default params:
params = {"sort": "userName-ASC", "size": 30, "page": 1}
Returns the changelog of the technical user with the given id
, including the creation of the user and the expiration extensions.
Default params:
params = {"sort": "createdAt-DESC", "size": 100, "page": 1}
Try to derive who is "responsible" for the technical user with the given id
. Usually, this is the user who created the technical user.
Creates a new user in LeanIX by creating an invitation (by default: silent invitation without e-mail).
This is required if you want to subscribe a user that has not been created in LeanIX yet.
role
can be "ADMIN"
, "MEMBER"
or "VIEWER"
.
To send out a real invitation, set silent=False
.
Note: Silent invitation without sending emails is not possible if LeanIX IDP is used - only for SSO workspaces.
Creates a new contact in LeanIX.
If the API token is not an ACCOUNTADMIN token, this method cannot create the contact directly. Instead, it
will use a workaround: The new user is subscribed to a dummy fact sheet, then removed from the fact sheet, and
then the fact sheet is deleted again.
You can pass a dummy fact sheet to this method using dummy_fs
. If no dummy_fs
is passed, the method will
create one and archive it at the end.
user
needs to be passed like this:
{
"email": "john.doe@example.com",
"firstName": "Firstname",
"lastName": "Lastname"
}
Creates a bookmark.
bookmark
is a dict describing the bookmark:
bookmark = {
"name": "My Saved Search",
"description": "My description",
"predefined": False,
"defaultSharingPriority": None,
"type": "INVENTORY",
"sharing": "SHARED",
"permittedReadUserIds": [],
"permittedWriteUserIds": [],
"state": {
"filter": {
"facetFilter": [],
"sorting": [{"key": "displayName", "order": "asc"}]
},
"viewMode": "list",
"tableConfig": {
"columns": [
{
"factSheetType": "Default",
"key": "displayName",
"sort": "",
"editable": False,
"sortable": True,
"resizable": True,
"type": "STRING"
},
{
"factSheetType": "Default",
"key": "type",
"sort": "",
"required": True,
"editable": False,
"sortable": True,
"width": 150,
"type": "STRING"
}
]
}
},
"workingCopy": None
}
Loads a specific bookmark, either by id or by name.
If a name is provided, the first matching bookmark is returned.
Runs the integration as specified by the ldif
parameter. Check test_factSheet.py
for an example.
Use wait=True to wait for the integration run to finish before returning, so that your automation knows when the next run can start without overloading the Integration API.
Use wait=False to run asynchronously.
Creates a relation between two fact sheets. Low-level function, only use if FactSheet function cannot be used.
Deletes the relation between two fact sheet types.
Low-level function, only use if FactSheet function cannot be used.
Loads ToDos. See documentation for FactSheet.todos()
for parameter details.
Creates a To-Do.
This is a low-level function that should only be used if you don't have a FactSheet instance where you can call FactSheet().create_todo(...).
fs_id is the fact sheet that the To-Do will be attached to.
user_ids is a list of user ids.
category is either "ACTION_ITEM" for ToDos, or "ANSWER" for questions.
due_date is a string in format YYYY-mm-dd.
Alternatively, due_days is a number of days, so that due_date is calculated automatically.
Example:
title = "My To-Do Item"
description = "My Description"
category = "ACTION_ITEM"
user_ids = [LeanIXUser.find_by_email("john.doe@example.com")["id"]]
due_date = "2029-12-31"
app.create_todo(fs_id="12345...", title=title, description=description, category=category, user_ids=user_ids, due_date=due_date)
See create_todo, but checks if a to-do in status OPEN or IN_PROGRESS with the same unique fields already exists, and updates the variable fields accordingly.
Deletes all ToDos with the given ids.
Example:
# Deletes all To-Dos of an application
app = Application("MyTest")
todos = app.todos()
LeanIXClient().delete_todos(ids=[item["id"] for item in todos])
Returns all fields that are not used in the current view config. Warning: Uses an unofficial API endpoint: /services/pathfinder/v1/models/metaModel
Downloads a resource to the local file system.
Example:
app = Application("My Test Application")
LeanIXClient().download_resource(app["documents"][0])
Creates and downloads a full snapshot.
filename
can contain the following format strings:
{time} = date + time
{workspace} = workspace name
Downloads the questions and results for each survey run as an Excel file.
filename
can contain the following format strings:
{survey} = survey id
{run} = run id
{time} = date + time
{workspace} = workspace name
Gets the permission object for a user, either by user_id
or by mail address.
Example:
permission = LeanIXClient().get_permission(mail="john.doe@example.com")
Updates a user's permission object, e.g. for changing a user's role from "VIEWER" to "MEMBER".
Example:
permission = LeanIXClient().get_permission(mail="john.doe@example.com")
permission["role"] = "MEMBER"
LeanIXClient().update_permission(permission)
Find the catalog items in the catalog catalog
(saas or ltls) matching to the query
.
Convenience method that takes a LeanIX webhook event as parameter and returns a dict that indicates: -
- which relations have been created
- which relations have been removed (i.e. deleted or archived)
- which relations have been updated
- which fact sheets are affected by this event because their relations or fields have been changed.
This is especially useful for translating the difficult RelationSwitch event - which is hard to interpret - into actionable sub-activities (one "create" and one "delete" activity).