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

Initial Sync

The Initial Synchronization allows to execute an automation for all fact sheets. Initial Synchronizations can be started in the web configuration menu for every automation that has the option supports_initial_sync = True.

The Initial Synchronization works as follows:

It simulates an event for every fact sheet of the fs_type that the automation is configured for
It executes the automation with the current configuration for every fact sheet
It shows a live log stream, displaying the log output for every execution of the handler

The default event that is passed to the event handler is a FactSheetUpdated event. If your automation relies on specific data within the event, your handler can implement a method called initial_sync_event, which takes the fact sheet as parameter and can build and return any event you like. For examples, look at some automations that implement the method "initial_sync_event", such as the FactSheetDeletionMailHandler.

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/

class Caching(enum.Enum):

on = <Caching.on: 'on'>

off = <Caching.off: 'off'>

class LeanIXClient:

Managing the connection to LeanIX and running all API calls.

LeanIXClient( api_token='AutoDetect', workspace_name=None, view=None, caching: Caching = <Caching.on: 'on'>)

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.

view

api_token

api_token_type

user_agent

caching

users

user_id

workspace_id

workspace_name

workspace_candidates

workspace

account_id

account

base_url

subscription_roles

tags

tag_groups

detected

user_type

user_name

user_name_first

user_name_last

user_name_display

customer_roles

access_control_entities

license_data

language_model

data_model

view_model

features

def set_cache(self, key, value):

def get_cache(self, key):

def get_cache_items(self):

def add_cache_run(self, name):

def has_cache_run(self, name):

def clear_cache(self):

def load_language_model(self, language='en', initial=False, force=False):

Loads the language model for language language into LeanIXClient().language_model[language].
Doesn't need to be called explicitly if you just need the English model; just access LeanIXClient().language_model and it will lazy-load the model.
Only used internally by FactSheet.field(), but can also be called explicitly if you want to load another language than English.

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 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, files=None, additional_headers=None):

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 mtm_graphql(self, query, variables=None, filter=None, additional_headers=None):

def graphql( self, query, variables=None, endpoint='/services/pathfinder/v1/graphql', filter=None, additional_headers=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 load_subscription_roles(self):

(Re-)loads all available subscription roles into self.subscription_roles.

Result:

{
  'Application Owner': {
    'id': 'd1c61fa4-050e-4713-9283-02f9ce43dded',
    'name': 'Application Owner',
    'description': '',
    'subscriptionType': 'RESPONSIBLE',
    'restrictToFactSheetTypes': [],
    'mandatoryForFactSheetTypes': []
  },
  'IT Manager': {
    ...
  }
}

def load_user_details(self):

Loads the user details of the API Token's user into self.user_name_[first|last|display]..

def create_subscription_role(self, role):

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.

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

Loads all available tag groups and their tags into self.tag_groups. Result looks like this:

{
    'Application Type': {
        'description': 'This tag group defines the application type.',
        'id': '02c7b01f-8927-4264-8c9d-4d3fe6df9b11',
        'mandatory': False,
        'mandatoryForFactSheetTypes': [],
        'mode': 'MULTIPLE',
        'name': 'Application Type',
        'restrictToFactSheetTypes': ['Application'],
        'shortName': None,
        'tagCount': 4,
        'tags': [
        {
                'color': '#75ff5b',
                'deletable': False,
                'description': 'A client application.',
                'factSheetCount': 3,
                'id': '4b45ead0-99dc-40f6-ab25-0f366ec09228',
                'name': 'Client',
                'status': 'ACTIVE'
        },
        {
                'color': '#5b6bff',
                'deletable': False,
                'description': 'Mobile Application',
                'factSheetCount': 14,
                'id': '13f7447c-c880-4278-ac1c-c7c3e319a1ae',
                'name': 'Mobile',
                'status': 'ACTIVE'
        }
    ]
}

def create_tag_group(self, tag_group):

Creates a new tag group with the given properties.

def create_tag(self, tag):

Creates a new tag with the given properties.

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

Deprecated, use get_webhooks() instead.

def get_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(self, webhook_id):

Deprecated, use get_webhook(webhook_id) instead.

def get_webhook_status(self, webhook_id):

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

def get_webhook_events(self, subscription_id):

Returns all events for this Pull Webhook.

def load_webhook_events(self, subscription_id):

Deprecated, use get_webhook_events(subscription_id) instead.

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, visibility='WORKSPACE'):

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"
visibility: "WORKSPACE" (default) or "PRIVATE"

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

Returns all Integrations.

def create_integration(self, integration):

Creates an Integration.

def create_or_update_integration(self, integration):

Creates or updates an Integration.

def delete_integration(self, integration):

Deletes an Integration.

def load_features(self):

Loads the LeanIX feature configuration for this workspace into LeanIXClient().features.

def get_feature(self, feature):

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

def get_quota(self):

Returns the quota (licensed applications) for this workspace.

def load_data_model(self):

Loads the workspace's data model into LeanIXClient().data_model. Doesn't need to be called explicitly; just access LeanIXClient().data_model and it will lazy-load the model.

def load_view_model(self):

Loads the workspace's view model into LeanIXClient().view_model. Doesn't need to be called explicitly; just access LeanIXClient().view_model and it will lazy-load the model.

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 update_bookmark(self, bookmark):

Updates a specific bookmark.

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

Loads all bookmarks.

def get_diagram(self, diagram_id=None, name=None):

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

def update_diagram(self, diagram):

Updates a diagram that was loaded via get_diagram().

def delete_diagram(self, diagram_id):

Deletes the diagram with the given id.

def run_integration(self, ldif, wait=True):

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.

def create_relation(self, source, target, rel_type, attributes=None):

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 get_todos(self, fs_ids=None, states=None, categories=None):

Returns all ToDos for the specified fact sheets. See documentation for FactSheet.todos() for parameter details.

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

Deprecated, use get_todos() instead.

def create_todo( self, fs_id, title, description, user_ids, category='ACTION_ITEM', due_date=None, due_days=None):

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)

def create_or_update_todo( self, fs_id, title, description, user_ids, category='ACTION_ITEM', due_date=None, due_days=None, unique_fields=None, variable_fields=None, recreate=False, mail_notification=None, fs=None, signal_create_callback=None):

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.

ToDos matching in all unique_fields will be deleted before creating new ToDos.
ToDos matching in all unique_fields AND all variable fields will be left unchanged if recreate is False.
Note: delete_outdated_todos() in quality_check_handler.py always uses ["title"], no other fields, so for consistency reasons this should be ["title"] here as well.

def update_todo(self, todo):

Updates a to-do.

def close_todo(self, todo_id):

Closes the to-do with the given todo_id.

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', language='en'):

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

def get_permissions(self):

Loads all permissions for this workspace.

def get_permission(self, mail=None, user_id=None):

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

def update_permission(self, permission):

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)

def find_catalog_items(self, catalog, query, fuzzy='false'):

Find the catalog items in the catalog catalog (saas or ltls) matching to the query.

@staticmethod

def get_event_info(event):

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

def get_sync_log(self):

Returns all LeanIX Sync Log entries.

def get_sync_log_items(self, id):

Returns the LeanIX Sync Log items for the Sync Log entry with the given id.

def get_sync_log_topics(self):

Returns all topics that are used in the LeanIX Sync Log.

def get_sync_log_triggers(self):

Returns all triggers that are used in the LeanIX Sync Log.

def create_sync_log_entry( self, name=None, created_at=None, finished_at=None, topic='LeanIX Automation Platform', trigger=None, message=None, scope='FULL', status='OK', progress='FINISHED', total=1, processed=0, warnings=0, errors=0, actions=0, items=None):

Creates a log entry in the LeanIX "Sync Logging" section.

status: "OK" / "WARNING" / "ERROR"
createdAt / finishedAt: will be auto-filled, no need to pass this value. progress: "RUNNING" / "FINISHED" / "FAILED"
scope: "PARTIAL" / "FULL"
items example:

[
    {
        "status": "OK",         # or "WARNING" or "ERROR"
        "createdAt": will be auto-filled, no need to pass this value.
        "action": "CREATE",     # or "UPDATE", "DELETE", "NONE"
        "sourceId": "My Source ID",
        "sourceName": "My Source Name",
        "sourceType": "My Source Type",
        "targetId": "My Target ID",
        "targetName": "My Target Name",
        "targetType": "My Target Type",
        "message": "This can be a long message."
    }
]

def update_sync_log_entry(self, id, entry):

Updates a LeanIX Sync Log entry.

def create_sync_log_entry_items(self, id, items=None):

Implicitly called by create_sync_log_entry() if the items parameter is specified.

def ingest_manifest(self, manifest_file):

Ingests a manifest file and creates the corresponding microservice fact sheet.

def ingest_sbom(self, fs_id, sbom_file):

Ingests an SBOM file into the microservice fact sheet with the given fact sheet ID.

def get_geo_locations(self, query, length=1):

Returns a list of length geo locations that match the given search query.

Example:

LeanIXClient().get_geo_locations(query="Zürich", length=1)

def get_transformations(self, fs_ids=None):

Returns all transformations for the specified fact sheet ids, or all transformations if no fs_ids are specified.

def get_all_action_batches(self):

Returns all meta model action batches for the current workspace.

def get_action_batchs(self, object_subtype, fs_type, name):

def get_actions_for_action_batch(self, batch_id):

Returns all meta model actions for the given action batch.

def create_action_batch(self, actions):

Creates new meta model actions.

def create_field( self, fs_type, field_name, field_type, section, subsection, index, size, weight, read_only, render_type, icon, values, metadata, translations, value_translations):

Creates a new meta model field. Parameters: fs_type: The fact sheet type, e.g. "Application" field_name: The name of the new field, usually in camelCase field_type: 'INTEGER', 'STRING', 'SINGLE_SELECT' or 'MULTI_SELECT' section: the ID of the section, as returned by Appliction.sections() (or other fact sheet types instead of Application) subsection: the ID of the subsection within the section above, as returned by Application.subsections(section=section) (or other fact sheet types instead of Application) index: The positional index of the new field within the subsection. Starts at 0. Cannot be larger than the length of Application.field_properties(section, subsection), which returns all current fields in the given subsection. size: 2 (XS), 3 (S), 4 (M), 6 (L), 10 (XL) or 12 (XXL). weight: a decimal for data quality completion weight, default is 1.0 read_only: True or False render_type: 'number', 'date', 'text', 'textarea'. For fields with type SINGLE_SELECT: 'status' icon: Only valid for SINGLE_SELECT fields. Example: {'color': '#000000', 'order': 'ASC', 'reverse': False, 'type': 'shield-alt'}. Type must be one of the following: star, question-circle, exclamation-square, exclamation-triangle, check-circle, badge-check, ban, do-not-enter, arrow-up, arrow-down, bell, eye, lock-alt, shield-alt, bug, gavel, hammer, toolbox, globe, atom, fire-alt, bolt, lemon, lightbulb, user-hard-hat, user, robot, chess-queen-alt, hat-wizard, unicorn, yen-sign, euro-sign, pound-sign, dollar-sign values: Only for SINGLE_SELECT and MULTI_SELECT fields. Example: ['level1', 'level2', 'level3', 'level4'] metadata: Instead of "icon" and "values", you can pass the metadata of an existing field, which supports more options, e.g. colors for individual values. translations: list of label translation items for the field itself. You need to pass translation items for each language you want to provide label translations for. Example item: [{ 'language': 'en', 'translation': { 'label': 'Recovery Time Objective', 'helpText': 'Help Text for RTO', 'textLabel': 'Recovery Time Objective', 'isLabelEmpty': False } }] value_translations: In case of SINGLE_SELECT and MULTI_SELECT fields, you also need to pass translations for the values. value_translations is a dict with every value as a key and a list of translation items as value: { "level1": [{ 'language': 'en', 'translation': { 'label': 'Low Criticality', 'helpText': 'Help Text for Low Criticality', 'textLabel': 'This is Low Criticality', 'isLabelEmpty': False }], "level2": [...] } Supported languages for translations - pass only the ones that you need: en, de, es, fr, pt, jp metadata example: { "values": { "administrativeService": { "bgColor": "#86d02f" }, "businessOperational": { "bgColor": "#fff300" }, "businessCritical": { "bgColor": "#fe8501" }, "missionCritical": { "bgColor": "#bf110d" } }, "icon": { "type": "arrow-up", "color": "#bf110d", "reverse": false, "order": "ASC" } }

def get_actions_for_field(self, fs_type, field):

def field_translations(self, fs_type, field):

Returns all translations and value translations for a field, so that they can be used for transporting or copying a field.

def get_actions_for_section(self, fs_type, section):

def get_actions_for_subsection(self, fs_type, subsection):

def create_section(self, fs_type, section, label):

def create_subsection(self, fs_type, section, subsection, label, help_text):

def get_calculations(self):

Loads all calculations in the workspace.

def get_calculation(self, calculation_id):

Loads the calculation with the given ID.

def create_calculation(self, calculation):

Creates the calculation specified by the calculation object.

    Example structure for simple calculation object:  

        {
          "name": "Draft Status longer than 30 days",
          "type": "fact-sheet",
          "affectedFactSheetType": "Application",
          "affectedFieldKey": "draftLongerThan30Days",
          "affectedFieldType": "SINGLE_SELECT",
          "description": "Calculates if draftDate is more than 30 days in the past",
          "status": "active",
          "code": "export function main() {
const draftDate = data.draftDate;

if (!draftDate) {
    return null;
}

const draftDateMs = new Date(draftDate).getTime();
const nowMs = new Date().getTime();
const diffInDays = Math.floor((nowMs - draftDateMs) / (1000 * 60 * 60 * 24));

if (diffInDays > 30) {
    return "yes";
} else {
    return "no";
}

}", "fields": [ { "type": "fact-sheet", "factSheetRelation": null, "factSheetRelationName": null, "factSheetRelationSide": null, "factSheetRelationCodeSide": null, "relation": null, "relationName": null, "relationSide": null, "triggerType": "change", "key": "draftDate" } ] }

    Example structure for calculation object with iteration across related fact sheets:  

          {
            "id": "a5505b8b-5419-4d9c-bd35-bd6c753bc465",
            "name": "Aggregate Foreign Attribute by Max/Min",
            "description": "",
            "active": true,
            "code": "export function main() {
let result = null;

for (const fs of data.relApplicationToProcess) {
    const input = fs.factSheet.recoveryPointObjective;
    if (input != null) {
        if (result == null || input < result) {     // for max instead of min, change "<" to ">"
            result = input;
        }
    }
}

return result;

} ", "language": "JAVASCRIPT", "version": 3, "updatedAt": "2025-08-16T08:14:27.125894Z", "createdAt": "2025-08-14T13:07:13.907312Z", "capability": { "type": "CALCULATIONS" } }

def update_calculation(self, calculation):

Updates the calculation specified by the calculation object.

def delete_calculation(self, calculation_id):

Deletes the calculation with the given ID.

def delete_calculations(self, interactive=True):

Careful: Deletes all calculations. interactive=True asks before deleting an calculation. To delete all calculations without asking, specify interactive=False.

def get_automations(self):

Loads all automations in the workspace.

def get_automation(self, automation_id):

Loads the automation with the given ID.

def create_automation(self, automation):

Creates the automation specified by the automation object.

def update_automation(self, automation):

Updates the automation specified by the automation object.

def delete_automation(self, automation_id):

Deletes the automation with the given ID.

def delete_automations(self, interactive=True):

Careful: Deletes all automations. interactive=True asks before deleting an automation. To delete all automations without asking, specify interactive=False.

def get_managed_code_execution_configurations(self):

Loads all managed code execution configurations in the workspace.

def get_managed_code_execution_configuration(self, config_id):

Loads the managed code execution configuration with the given ID.

def get_managed_code_execution_configuration_log_entries(self, config_id):

Loads all log entries of the managed code execution configuration with the given ID.

def get_managed_code_execution_configuration_log_entry(self, config_id, log_id):

Loads a single log entry of the managed code execution configuration log entry with the given ID.

def create_managed_code_execution_configuration(self, config):

Creates the managed code execution configuration specified by the automation object.

def update_managed_code_execution_configuration(self, config):

Updates the managed code execution configuration specified by the configuration object.

def delete_managed_code_execution_configuration(self, config_id):

Deletes the managed code execution configuration with the given ID. Not available in LeanIX yet.

def delete_managed_code_execution_configurations(self, interactive=True):

Careful: Deletes all managed code execution configurations. interactive=True asks before deleting a configuration. To delete all configurations without asking, specify interactive=False.

def transport_integrations(self, integrations, target_workspace_api_token, interactive=True):

def transport_webhooks(self, webhooks, target_workspace_api_token, interactive=True):

def transport_subscription_roles(self, roles, target_workspace_api_token, interactive=True):

def transport_automations(self, automations, target_workspace_api_token, interactive=True):

def transport_calculations(self, calculations, target_workspace_api_token, interactive=True):

def transport_managed_code_executions( self, managed_code_executions, target_workspace_api_token, interactive=True):

def transport_tags(self, elements, target_workspace_api_token, interactive=True):

def transport_meta_model_elements( self, meta_model_elements, target_workspace_api_token, source_fs_type, target_fs_type, interactive=True):

def transport_objects( self, object_type, target_workspace_api_token, object_subtype=None, source_fs_type=None, target_fs_type=None, object_list=None, interactive=True):