utils.leanix_client

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 sheet
  • fact_sheet_deletion_mail_handler.FactSheetDeletionMailHandler to receive an e-mail notification when somebody archives a fact sheet
  • subscription_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 FactSheets and organization.Organizations, 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).

class LeanIXClient:

Managing the connection to LeanIX and running all API calls.

LeanIXClient(api_token='AutoDetect', workspace_name=None)

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.

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 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"],  
]
further_tag_sets = [['pathfinder', 'DOCUMENT_UPDATED'], ['pathfinder', 'DOCUMENT_DELETED'], ['pathfinder', 'FACT_SHEET_RECOVERED'], ['pathfinder', 'COMMENT_CREATED'], ['pathfinder', 'COMMENT_DELETED'], ['pathfinder', 'BOOKMARK_CREATED'], ['pathfinder', 'BOOKMARK_UPDATED'], ['pathfinder', 'BOOKMARK_DELETED'], ['poll', 'POLL_CREATED'], ['poll', 'POLL_FINISHED'], ['pollResult', 'POLL_RESULT_UPDATED'], ['pathfinder', 'SURVEY_CREATED'], ['pathfinder', 'SURVEY_UPDATED'], ['pathfinder', 'SURVEY_DELETED'], ['pathfinder', 'SURVEY_RUN_CREATED'], ['pathfinder', 'SURVEY_RUN_DELETED'], ['pathfinder', 'WEBHOOK_CREATED'], ['pathfinder', 'WEBHOOK_UPDATED'], ['pathfinder', 'WEBHOOK_DELETED'], ['pathfinder', 'QUALITY_SEAL_APPROVED'], ['pathfinder', 'QUALITY_SEAL_BROKEN'], ['pathfinder', 'RELATION_VALIDITY_FROM_CHANGED'], ['pathfinder', 'RELATION_VALIDITY_UNTIL_CHANGED'], ['pathfinder', 'RELATION_RECOVERED'], [['integrations', 'INTEGRATION-API'], 'statistics']]

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.

def load_language_model(self, language):

Only used internally by FactSheet.field().
Loads the language model for language language into LeanIXClient().language_model[language].

def load_users(self, options=None):

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.

def get_workspace_url(self):

Returns the workspace URL, which is used to generate the links in the automatic e-mails.

def connect(self):

Connects to LeanIX and acquires an oauth2 access token.

def detect_workspace_properties(self):

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.

def request( self, path, method, body=None, params=None, options=None, stream=False):

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.

def graphql( self, query, variables=None, endpoint='/services/pathfinder/v1/graphql', filter=None):

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().

def get_subscription_roles(self):

Loads all available subscription roles into self.subscription_roles.

def create_subscription_role(self, subscription_type, name):

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().

def get_tag_group_id(self, tag_group):

Returns the tag group id for a specific tag group name.

def get_tag_id(self, tag_group, tag):

Returns the tag id for a specific tag name and tag group name.

def delete_tag(self, tag_id):

Warning: This deletes a tag from the workspace configuration. If you want to delete a tag from a fact sheet, use FactSheet.delete_tags().

def delete_tag_group(self, tag_group_id):

Warning: This deletes a tag from the workspace configuration. If you want to delete a tag from a fact sheet, use FactSheet.delete_tags().

def get_role_id(self, role_name):

Returns the id for a specific subscription role name.

@classmethod
def get_query_extension_by_field(cls, field):
def get_tags(self):

Loads all available tags including their IDs into self.tags, so that they can be used for name-based tagging without tag IDs.

self.tags looks like this:

self.tags = {
    "Tag Group 1": {
        "Tag 1": {"id": "TAG_ID1", "description": "tag description"},
        "Tag 2": {"id": "TAG_ID2", "description": "tag description"},
        "Tag 3": {"id": "TAG_ID3", "description": "tag description"},
    },
    "Tag Group 2": ...
}
def get_tag_groups(self):

Returns all available tag groups.

def create_tag_group( self, name, shortName, description, mode, restrictToFactSheetTypes, mandatory=False):

Creates a new tag group with the given properties.

def create_tag(self, name, tagGroup, color, description):

Creates a new tag with the given properties.

def load_webhooks(self):

Returns all Webhooks subscriptions.
Note: If the Managing User of a Webhook is set to "Current User (private Webhook)", other users cannot see it.

def load_webhook(self, webhook_id):

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.

def load_webhook_events(self, subscription_id):

Returns all events for this Pull Webhook.

def advance_webhook_cursor(self, subscription_id, cursor):

Advances a Pull Webhook's cursor after an event has been processed successfully.

def get_webhook_logs(self, webhook_id, params=None, options=None):

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})`
def recreate_webhook(self, webhook_id):

Sometimes LeanIX breaks webhooks. This function fixes a broken webhook by deleting and recreating it. Only works for PUSH webhooks at the moment.

def create_webhook(self, wh):

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().

def fix_webhook_event_types(self, id=None):

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.

def register_webhook( self, name, target, user, pwd, delivery_type, active=True, ignore_error=False):

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).

def update_webhook(self, webhook):

Updates a Webhook.

webhook is the dict structure containing the webhook data, since it needs to be passed during the update.

def delete_webhook(self, webhook_id):

Deletes the webhook with the given id.

def add_webhook_event_type(self, webhook_id, event_type, tags='pathfinder'):

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")
def edit_webhook_interactively(self):

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()
def get_features(self):

Returns the LeanIX feature configuration for this workspace.

def get_feature(self, feature):

Returns the configuration of a specific feature for this workspace. See get_features() for full list.

def create_schema(self, schema):

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)
def get_schemas(self):

Loads all metrics schemas.

def get_schema(self, name=None, id=None):

Loads a metrics schema by its id or name.

def delete_schema(self, id):

Deletes a schema by its id.

def get_metric_points(self, id, options=None):

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})
def get_metric_point(self, id, timestamp):

Gets one single metric point by schema id and timestamp. See get_metric_points for timestamp requirements.

def post_metric_point(self, id, point):

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)
def post_metric_points(self, id, points):

Calls post_metric_point for every single point in the points list.

def delete_metric_point(self, id, timestamp):

Deletes the metric point with the given timestamp from the schema with the given id. See get_metric_points for timestamp requirements.

def delete_metric_points(self, id):

Deletes all metric points from the given schema id.

def delete_metric_points_range(self, id, begin, end):

Deletes all metric points within a range.
Currently not supported properly by LeanIX - returns errors (2023-01-15).

def get_workspace_settings(self):

Returns the workspace settings object as JSON.

def backup_workspace_settings(self):

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.

def update_workspace_settings(self, settings):

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.

def create_chart(self, new_chart):

Creates a single metrics chart. Currently only used internally for the Data Quality Management report.

def create_kpi_charts(self):

Creates all KPI charts for the Data Quality Management report. Only used internally at the moment.

def get_technicalusers(self, params=None):

Returns all technical users.
Default params:

params = {"sort": "userName-ASC", "size": 30, "page": 1}
def get_technicaluser_changelog(self, id, params=None):

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}
def get_responsible_user_for_technical_user(self, id):

Try to derive who is "responsible" for the technical user with the given id. Usually, this is the user who created the technical user.

def create_user(self, mail, role='MEMBER', silent=True):

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.

def create_contact(self, user, dummy_fs=None):

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"
}
def create_bookmark(self, bookmark=None):

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
}
def delete_bookmark(self, bookmark_id):

Deletes the bookmark with the given id.

def get_bookmark(self, bookmark_id=None, name=None):

Loads a specific bookmark, either by id or by name.
If a name is provided, the first matching bookmark is returned.

def get_bookmarks(self, bookmark_type='REPORTING'):

Loads all bookmarks.

def run_integration(self, ldif):

Runs the integration as specified by the ldif parameter. Check test_factSheet.py for an example.

def create_relation(self, source, target, rel_type):

Creates a relation between two fact sheets. Low-level function, only use if FactSheet function cannot be used.

def delete_relation(self, source, target, rel_id, rel_type):

Deletes the relation between two fact sheet types.
Low-level function, only use if FactSheet function cannot be used.

def load_todos(self, fs_ids=None, states=None):

Loads ToDos. See documentation for FactSheet.load_todos() for parameter details.

def delete_todos(self, ids):

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])
def chat(self, message, factsheet, context):
def unused_fields(self):

Returns all fields that are not used in the current view config. Warning: Uses an unofficial API endpoint: /services/pathfinder/v1/models/metaModel

def get_acls(self):

Returns the workspace's ACL configuration.

def get_authorizations(self):

Returns the workspace's authorizations and permission settings.

def export_authorizations(self, file_name='permissions.csv'):

Exports all authorizations into an Excel/CSV format.
Warning: Uses an unofficial API - not guaranteed to remain functional.

def download_resource(self, document):

Downloads a resource to the local file system.

Example:

app = Application("My Test Application")
LeanIXClient().download_resource(app["documents"][0])
def download_snapshot( self, filename='full_snapshot_{workspace}_{time}.xlsx', timeout_secs=14400):

Creates and downloads a full snapshot.
filename can contain the following format strings:
{time} = date + time
{workspace} = workspace name

def download_surveys(self, filename='survey_run_{workspace}_{time}_{survey}_{run}.xlsx'):

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