lusid_workflow.WorkflowsApi

All URIs are relative to https://fbn-prd.lusid.com/workflow

Method	HTTP request	Description
create_workflow	POST /api/workflows	CreateWorkflow: Create a new Workflow
delete_workflow	DELETE /api/workflows/{scope}/{code}	[EXPERIMENTAL] DeleteWorkflow: Delete a Workflow
get_workflow	GET /api/workflows/{scope}/{code}	GetWorkflow: Get a Workflow
list_workflows	GET /api/workflows	ListWorkflows: List Workflows
update_workflow	PUT /api/workflows/{scope}/{code}	[EXPERIMENTAL] UpdateWorkflow: Update an existing Workflow
upsert_workflow_properties	POST /api/workflows/{scope}/{code}/properties	[EXPERIMENTAL] UpsertWorkflowProperties: Add, update and remove properties on an existing Workflow in bulk.

create_workflow

WorkflowResponse create_workflow(create_workflow_request)

CreateWorkflow: Create a new Workflow

Example

from lusid_workflow.exceptions import ApiException
from lusid_workflow.extensions.configuration_options import ConfigurationOptions
from lusid_workflow.models import *
from pprint import pprint
from lusid_workflow import (
    SyncApiClientFactory,
    WorkflowsApi
)

def main():

    with open("secrets.json", "w") as file:
        file.write('''
    {
        "api":
        {
            "tokenUrl":"<your-token-url>",
            "workflowUrl":"https://<your-domain>.lusid.com/workflow",
            "username":"<your-username>",
            "password":"<your-password>",
            "clientId":"<your-client-id>",
            "clientSecret":"<your-client-secret>"
        }
    }''')

    # Use the lusid_workflow SyncApiClientFactory to build Api instances with a configured api client
    # By default this will read config from environment variables
    # Then from a secrets.json file found in the current working directory

    # uncomment the below to use configuration overrides
    # opts = ConfigurationOptions();
    # opts.total_timeout_ms = 30_000

    # uncomment the below to use an api client factory with overrides
    # api_client_factory = SyncApiClientFactory(opts=opts)

    api_client_factory = SyncApiClientFactory()

    # Enter a context with an instance of the SyncApiClientFactory to ensure the connection pool is closed after use
    
    # Create an instance of the API class
    api_instance = api_client_factory.build(WorkflowsApi)

    # Objects can be created either via the class constructor, or using the 'from_dict' or 'from_json' methods
    # Change the lines below to switch approach
    # create_workflow_request = CreateWorkflowRequest.from_json("")
    # create_workflow_request = CreateWorkflowRequest.from_dict({})
    create_workflow_request = CreateWorkflowRequest()

    try:
        # uncomment the below to set overrides at the request level
        # api_response =  api_instance.create_workflow(create_workflow_request, opts=opts)

        # CreateWorkflow: Create a new Workflow
        api_response = api_instance.create_workflow(create_workflow_request)
        pprint(api_response)

    except ApiException as e:
        print("Exception when calling WorkflowsApi->create_workflow: %s\n" % e)

main()

Parameters

Name	Type	Description	Notes
create_workflow_request	CreateWorkflowRequest	The data to create a Workflow

Return type

WorkflowResponse

HTTP request headers

Content-Type: application/json-patch+json, application/json, text/json, application/*+json
Accept: application/json

HTTP response details

Status code	Description	Response headers
201	Created	-
400	The details of the input related failure	-
409	Workflow already exists.	-
0	Error response	-

delete_workflow

DeletedEntityResponse delete_workflow(scope, code)

[EXPERIMENTAL] DeleteWorkflow: Delete a Workflow

Example

from lusid_workflow.exceptions import ApiException
from lusid_workflow.extensions.configuration_options import ConfigurationOptions
from lusid_workflow.models import *
from pprint import pprint
from lusid_workflow import (
    SyncApiClientFactory,
    WorkflowsApi
)

def main():

    with open("secrets.json", "w") as file:
        file.write('''
    {
        "api":
        {
            "tokenUrl":"<your-token-url>",
            "workflowUrl":"https://<your-domain>.lusid.com/workflow",
            "username":"<your-username>",
            "password":"<your-password>",
            "clientId":"<your-client-id>",
            "clientSecret":"<your-client-secret>"
        }
    }''')

    # Use the lusid_workflow SyncApiClientFactory to build Api instances with a configured api client
    # By default this will read config from environment variables
    # Then from a secrets.json file found in the current working directory

    # uncomment the below to use configuration overrides
    # opts = ConfigurationOptions();
    # opts.total_timeout_ms = 30_000

    # uncomment the below to use an api client factory with overrides
    # api_client_factory = SyncApiClientFactory(opts=opts)

    api_client_factory = SyncApiClientFactory()

    # Enter a context with an instance of the SyncApiClientFactory to ensure the connection pool is closed after use
    
    # Create an instance of the API class
    api_instance = api_client_factory.build(WorkflowsApi)
    scope = 'scope_example' # str | The scope that identifies a Workflow
    code = 'code_example' # str | The code that identifies a Workflow

    try:
        # uncomment the below to set overrides at the request level
        # api_response =  api_instance.delete_workflow(scope, code, opts=opts)

        # [EXPERIMENTAL] DeleteWorkflow: Delete a Workflow
        api_response = api_instance.delete_workflow(scope, code)
        pprint(api_response)

    except ApiException as e:
        print("Exception when calling WorkflowsApi->delete_workflow: %s\n" % e)

main()

Parameters

Name	Type	Description	Notes
scope	str	The scope that identifies a Workflow
code	str	The code that identifies a Workflow

Return type

DeletedEntityResponse

HTTP request headers

Content-Type: Not defined
Accept: application/json

HTTP response details

Status code	Description	Response headers
200	OK	-
400	The details of the input related failure	-
404	Workflow not found.	-
0	Error response	-

get_workflow

WorkflowResponse get_workflow(scope, code, as_at=as_at, property_keys=property_keys)

GetWorkflow: Get a Workflow

Example

from lusid_workflow.exceptions import ApiException
from lusid_workflow.extensions.configuration_options import ConfigurationOptions
from lusid_workflow.models import *
from pprint import pprint
from lusid_workflow import (
    SyncApiClientFactory,
    WorkflowsApi
)

def main():

    with open("secrets.json", "w") as file:
        file.write('''
    {
        "api":
        {
            "tokenUrl":"<your-token-url>",
            "workflowUrl":"https://<your-domain>.lusid.com/workflow",
            "username":"<your-username>",
            "password":"<your-password>",
            "clientId":"<your-client-id>",
            "clientSecret":"<your-client-secret>"
        }
    }''')

    # Use the lusid_workflow SyncApiClientFactory to build Api instances with a configured api client
    # By default this will read config from environment variables
    # Then from a secrets.json file found in the current working directory

    # uncomment the below to use configuration overrides
    # opts = ConfigurationOptions();
    # opts.total_timeout_ms = 30_000

    # uncomment the below to use an api client factory with overrides
    # api_client_factory = SyncApiClientFactory(opts=opts)

    api_client_factory = SyncApiClientFactory()

    # Enter a context with an instance of the SyncApiClientFactory to ensure the connection pool is closed after use
    
    # Create an instance of the API class
    api_instance = api_client_factory.build(WorkflowsApi)
    scope = 'scope_example' # str | The scope that identifies a Workflow
    code = 'code_example' # str | The code that identifies a Workflow
    as_at = '2013-10-20T19:20:30+01:00' # datetime | The asAt datetime at which to retrieve the Workflow. Defaults to returning the latest version if not specified. (optional)
    property_keys = ['property_keys_example'] # List[str] | The property keys (in the Workflow or TaskDefinition domain) whose values to return on the Workflow. (optional)

    try:
        # uncomment the below to set overrides at the request level
        # api_response =  api_instance.get_workflow(scope, code, as_at=as_at, property_keys=property_keys, opts=opts)

        # GetWorkflow: Get a Workflow
        api_response = api_instance.get_workflow(scope, code, as_at=as_at, property_keys=property_keys)
        pprint(api_response)

    except ApiException as e:
        print("Exception when calling WorkflowsApi->get_workflow: %s\n" % e)

main()

Parameters

Name	Type	Description	Notes
scope	str	The scope that identifies a Workflow
code	str	The code that identifies a Workflow
as_at	datetime	The asAt datetime at which to retrieve the Workflow. Defaults to returning the latest version if not specified.	[optional]
property_keys	List[str]	The property keys (in the Workflow or TaskDefinition domain) whose values to return on the Workflow.	[optional]

Return type

WorkflowResponse

HTTP request headers

Content-Type: Not defined
Accept: application/json

HTTP response details

Status code	Description	Response headers
200	OK	-
400	The details of the input related failure	-
404	Workflow not found.	-
0	Error response	-

list_workflows

PagedResourceListOfWorkflowResponse list_workflows(as_at=as_at, filter=filter, sort_by=sort_by, limit=limit, page=page, property_keys=property_keys)

ListWorkflows: List Workflows

Example

from lusid_workflow.exceptions import ApiException
from lusid_workflow.extensions.configuration_options import ConfigurationOptions
from lusid_workflow.models import *
from pprint import pprint
from lusid_workflow import (
    SyncApiClientFactory,
    WorkflowsApi
)

def main():

    with open("secrets.json", "w") as file:
        file.write('''
    {
        "api":
        {
            "tokenUrl":"<your-token-url>",
            "workflowUrl":"https://<your-domain>.lusid.com/workflow",
            "username":"<your-username>",
            "password":"<your-password>",
            "clientId":"<your-client-id>",
            "clientSecret":"<your-client-secret>"
        }
    }''')

    # Use the lusid_workflow SyncApiClientFactory to build Api instances with a configured api client
    # By default this will read config from environment variables
    # Then from a secrets.json file found in the current working directory

    # uncomment the below to use configuration overrides
    # opts = ConfigurationOptions();
    # opts.total_timeout_ms = 30_000

    # uncomment the below to use an api client factory with overrides
    # api_client_factory = SyncApiClientFactory(opts=opts)

    api_client_factory = SyncApiClientFactory()

    # Enter a context with an instance of the SyncApiClientFactory to ensure the connection pool is closed after use
    
    # Create an instance of the API class
    api_instance = api_client_factory.build(WorkflowsApi)
    as_at = '2013-10-20T19:20:30+01:00' # datetime | The asAt datetime at which to list the Workflows. Defaults to return the latest version of each Workflow if not specified. (optional)
    filter = 'filter_example' # str | Expression to filter the result set. Read more about filtering results from LUSID here: https://support.lusid.com/filtering-results-from-lusid. (optional)
    sort_by = ['sort_by_example'] # List[str] | A list of field names or properties to sort by, each suffixed by \" ASC\" or \" DESC\" (optional)
    limit = 10 # int | When paginating, limit the number of returned results to this many. (optional) (default to 10)
    page = 'page_example' # str | The pagination token to use to continue listing workflows from a previous call to list workflows. This value is returned from the previous call. If a pagination token is provided the sortBy, filter, effectiveAt, and asAt fields must not have changed since the original request. (optional)
    property_keys = ['property_keys_example'] # List[str] | The property keys (in the Workflow or TaskDefinition domain) whose values to return on each Workflow. (optional)

    try:
        # uncomment the below to set overrides at the request level
        # api_response =  api_instance.list_workflows(as_at=as_at, filter=filter, sort_by=sort_by, limit=limit, page=page, property_keys=property_keys, opts=opts)

        # ListWorkflows: List Workflows
        api_response = api_instance.list_workflows(as_at=as_at, filter=filter, sort_by=sort_by, limit=limit, page=page, property_keys=property_keys)
        pprint(api_response)

    except ApiException as e:
        print("Exception when calling WorkflowsApi->list_workflows: %s\n" % e)

main()

Parameters

Name	Type	Description	Notes
as_at	datetime	The asAt datetime at which to list the Workflows. Defaults to return the latest version of each Workflow if not specified.	[optional]
filter	str	Expression to filter the result set. Read more about filtering results from LUSID here: https://support.lusid.com/filtering-results-from-lusid.	[optional]
sort_by	List[str]	A list of field names or properties to sort by, each suffixed by " ASC" or " DESC"	[optional]
limit	int	When paginating, limit the number of returned results to this many.	[optional] [default to 10]
page	str	The pagination token to use to continue listing workflows from a previous call to list workflows. This value is returned from the previous call. If a pagination token is provided the sortBy, filter, effectiveAt, and asAt fields must not have changed since the original request.	[optional]
property_keys	List[str]	The property keys (in the Workflow or TaskDefinition domain) whose values to return on each Workflow.	[optional]

Return type

PagedResourceListOfWorkflowResponse

HTTP request headers

Content-Type: Not defined
Accept: application/json

HTTP response details

Status code	Description	Response headers
200	OK	-
400	The details of the input related failure	-
404	No Workflows found.	-
0	Error response	-

update_workflow

WorkflowResponse update_workflow(scope, code, update_workflow_request)

[EXPERIMENTAL] UpdateWorkflow: Update an existing Workflow

Example

from lusid_workflow.exceptions import ApiException
from lusid_workflow.extensions.configuration_options import ConfigurationOptions
from lusid_workflow.models import *
from pprint import pprint
from lusid_workflow import (
    SyncApiClientFactory,
    WorkflowsApi
)

def main():

    with open("secrets.json", "w") as file:
        file.write('''
    {
        "api":
        {
            "tokenUrl":"<your-token-url>",
            "workflowUrl":"https://<your-domain>.lusid.com/workflow",
            "username":"<your-username>",
            "password":"<your-password>",
            "clientId":"<your-client-id>",
            "clientSecret":"<your-client-secret>"
        }
    }''')

    # Use the lusid_workflow SyncApiClientFactory to build Api instances with a configured api client
    # By default this will read config from environment variables
    # Then from a secrets.json file found in the current working directory

    # uncomment the below to use configuration overrides
    # opts = ConfigurationOptions();
    # opts.total_timeout_ms = 30_000

    # uncomment the below to use an api client factory with overrides
    # api_client_factory = SyncApiClientFactory(opts=opts)

    api_client_factory = SyncApiClientFactory()

    # Enter a context with an instance of the SyncApiClientFactory to ensure the connection pool is closed after use
    
    # Create an instance of the API class
    api_instance = api_client_factory.build(WorkflowsApi)
    scope = 'scope_example' # str | The scope that identifies a Workflow
    code = 'code_example' # str | The code that identifies a Workflow

    # Objects can be created either via the class constructor, or using the 'from_dict' or 'from_json' methods
    # Change the lines below to switch approach
    # update_workflow_request = UpdateWorkflowRequest.from_json("")
    # update_workflow_request = UpdateWorkflowRequest.from_dict({})
    update_workflow_request = UpdateWorkflowRequest()

    try:
        # uncomment the below to set overrides at the request level
        # api_response =  api_instance.update_workflow(scope, code, update_workflow_request, opts=opts)

        # [EXPERIMENTAL] UpdateWorkflow: Update an existing Workflow
        api_response = api_instance.update_workflow(scope, code, update_workflow_request)
        pprint(api_response)

    except ApiException as e:
        print("Exception when calling WorkflowsApi->update_workflow: %s\n" % e)

main()

Parameters

Name	Type	Description
scope	str	The scope that identifies a Workflow
code	str	The code that identifies a Workflow
update_workflow_request	UpdateWorkflowRequest	The data to update a Workflow

Return type

WorkflowResponse

HTTP request headers

Content-Type: application/json-patch+json, application/json, text/json, application/*+json
Accept: application/json

HTTP response details

Status code	Description	Response headers
200	OK	-
400	The details of the input related failure	-
404	Workflow not found.	-
0	Error response	-

upsert_workflow_properties

BatchUpsertWorkflowPropertiesResponse upsert_workflow_properties(scope, code, request_body, success_mode=success_mode)

[EXPERIMENTAL] UpsertWorkflowProperties: Add, update and remove properties on an existing Workflow in bulk.

Example

from lusid_workflow.exceptions import ApiException
from lusid_workflow.extensions.configuration_options import ConfigurationOptions
from lusid_workflow.models import *
from pprint import pprint
from lusid_workflow import (
    SyncApiClientFactory,
    WorkflowsApi
)

def main():

    with open("secrets.json", "w") as file:
        file.write('''
    {
        "api":
        {
            "tokenUrl":"<your-token-url>",
            "workflowUrl":"https://<your-domain>.lusid.com/workflow",
            "username":"<your-username>",
            "password":"<your-password>",
            "clientId":"<your-client-id>",
            "clientSecret":"<your-client-secret>"
        }
    }''')

    # Use the lusid_workflow SyncApiClientFactory to build Api instances with a configured api client
    # By default this will read config from environment variables
    # Then from a secrets.json file found in the current working directory

    # uncomment the below to use configuration overrides
    # opts = ConfigurationOptions();
    # opts.total_timeout_ms = 30_000

    # uncomment the below to use an api client factory with overrides
    # api_client_factory = SyncApiClientFactory(opts=opts)

    api_client_factory = SyncApiClientFactory()

    # Enter a context with an instance of the SyncApiClientFactory to ensure the connection pool is closed after use
    
    # Create an instance of the API class
    api_instance = api_client_factory.build(WorkflowsApi)
    scope = 'scope_example' # str | The scope that identifies a Workflow
    code = 'code_example' # str | The code that identifies a Workflow
    request_body = {"Workflow/myScope/reviewer":{"key":"Workflow/myScope/reviewer","value":{"labelValue":"Jane Doe"}},"Workflow/myScope/priority":{"key":"Workflow/myScope/priority","value":{"metricValue":{"value":1}}},"Workflow/myScope/obsoleteProperty":{"key":"Workflow/myScope/obsoleteProperty"}} # Dict[str, PerpetualProperty] | The properties to upsert, keyed by property key. A null value deletes the property.
    success_mode = 'Partial' # str | Whether the batch should fail Atomically or Partially. Defaults to Partial. (optional) (default to 'Partial')

    try:
        # uncomment the below to set overrides at the request level
        # api_response =  api_instance.upsert_workflow_properties(scope, code, request_body, success_mode=success_mode, opts=opts)

        # [EXPERIMENTAL] UpsertWorkflowProperties: Add, update and remove properties on an existing Workflow in bulk.
        api_response = api_instance.upsert_workflow_properties(scope, code, request_body, success_mode=success_mode)
        pprint(api_response)

    except ApiException as e:
        print("Exception when calling WorkflowsApi->upsert_workflow_properties: %s\n" % e)

main()

Parameters

Name	Type	Description	Notes
scope	str	The scope that identifies a Workflow
code	str	The code that identifies a Workflow
request_body	Dict[str, PerpetualProperty]	The properties to upsert, keyed by property key. A null value deletes the property.
success_mode	str	Whether the batch should fail Atomically or Partially. Defaults to Partial.	[optional] [default to 'Partial']

Return type

BatchUpsertWorkflowPropertiesResponse

HTTP request headers

Content-Type: application/json-patch+json, application/json, text/json, application/*+json
Accept: application/json

HTTP response details

Status code	Description	Response headers
200	OK	-
400	The details of the input related failure	-
404	Workflow not found.	-
0	Error response	-

FilesExpand file tree

WorkflowsApi.md

Latest commit

History

WorkflowsApi.md

File metadata and controls

lusid_workflow.WorkflowsApi

create_workflow

Example

Parameters

Return type

HTTP request headers

HTTP response details

delete_workflow

Example

Parameters

Return type

HTTP request headers

HTTP response details

get_workflow

Example

Parameters

Return type

HTTP request headers

HTTP response details

list_workflows

Example

Parameters

Return type

HTTP request headers

HTTP response details

update_workflow

Example

Parameters

Return type

HTTP request headers

HTTP response details

upsert_workflow_properties

Example

Parameters

Return type

HTTP request headers

HTTP response details