lusid_workflow.TaskDefinitionsApi

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

Method	HTTP request	Description
create_task_definition	POST /api/taskdefinitions	CreateTaskDefinition: Create a new Task Definition
delete_task_definition	DELETE /api/taskdefinitions/{scope}/{code}	DeleteTaskDefinition: Delete a Task Definition
get_task_definition	GET /api/taskdefinitions/{scope}/{code}	GetTaskDefinition: Get a Task Definition
list_task_definitions	GET /api/taskdefinitions	ListTaskDefinitions: List Task Definitions
list_tasks_for_task_definition	GET /api/taskdefinitions/{scope}/{code}/tasks	ListTasksForTaskDefinition: List Tasks for a Task Definition
update_task_definition	PUT /api/taskdefinitions/{scope}/{code}	UpdateTaskDefinition: Update an existing Task Definition
upsert_task_definition_properties	POST /api/taskdefinitions/{scope}/{code}/properties	[EXPERIMENTAL] UpsertTaskDefinitionProperties: Add, update and remove properties on an existing Task Definition in bulk.

create_task_definition

TaskDefinition create_task_definition(create_task_definition_request)

CreateTaskDefinition: Create a new Task Definition

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,
    TaskDefinitionsApi
)

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

    # 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_task_definition_request = CreateTaskDefinitionRequest.from_json("")
    # create_task_definition_request = CreateTaskDefinitionRequest.from_dict({})
    create_task_definition_request = CreateTaskDefinitionRequest()

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

        # CreateTaskDefinition: Create a new Task Definition
        api_response = api_instance.create_task_definition(create_task_definition_request)
        pprint(api_response)

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

main()

Parameters

Name	Type	Description	Notes
create_task_definition_request	CreateTaskDefinitionRequest	The data to create a Task Definition

Return type

TaskDefinition

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	-
0	Error response	-

delete_task_definition

DeletedEntityResponse delete_task_definition(scope, code)

DeleteTaskDefinition: Delete a Task Definition

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,
    TaskDefinitionsApi
)

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(TaskDefinitionsApi)
    scope = 'scope_example' # str | The scope that identifies a Task Definition
    code = 'code_example' # str | The code that identifies a Task Definition

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

        # DeleteTaskDefinition: Delete a Task Definition
        api_response = api_instance.delete_task_definition(scope, code)
        pprint(api_response)

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

main()

Parameters

Name	Type	Description	Notes
scope	str	The scope that identifies a Task Definition
code	str	The code that identifies a Task Definition

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	Task Definition not found.	-
0	Error response	-

get_task_definition

TaskDefinition get_task_definition(scope, code, as_at=as_at, property_keys=property_keys)

GetTaskDefinition: Get a Task Definition

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,
    TaskDefinitionsApi
)

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(TaskDefinitionsApi)
    scope = 'scope_example' # str | The scope that identifies a Task Definition
    code = 'code_example' # str | The code that identifies a Task Definition
    as_at = '2013-10-20T19:20:30+01:00' # datetime | The asAt datetime at which to retrieve the Task Definition. Defaults to returning the latest version of the Task Definition if not specified. (optional)
    property_keys = ['property_keys_example'] # List[str] | The property keys whose values to return on the Task Definition. (optional)

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

        # GetTaskDefinition: Get a Task Definition
        api_response = api_instance.get_task_definition(scope, code, as_at=as_at, property_keys=property_keys)
        pprint(api_response)

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

main()

Parameters

Name	Type	Description	Notes
scope	str	The scope that identifies a Task Definition
code	str	The code that identifies a Task Definition
as_at	datetime	The asAt datetime at which to retrieve the Task Definition. Defaults to returning the latest version of the Task Definition if not specified.	[optional]
property_keys	List[str]	The property keys whose values to return on the Task Definition.	[optional]

Return type

TaskDefinition

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	Task Definition not found.	-
0	Error response	-

list_task_definitions

PagedResourceListOfTaskDefinition list_task_definitions(as_at=as_at, filter=filter, sort_by=sort_by, limit=limit, page=page, property_keys=property_keys)

ListTaskDefinitions: List Task Definitions

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,
    TaskDefinitionsApi
)

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(TaskDefinitionsApi)
    as_at = '2013-10-20T19:20:30+01:00' # datetime | The asAt datetime at which to list the Task Definitions. Defaults to return the latest version of each Task Definition 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 task definitions from a previous call to list task definitions. 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 whose values to return on each Task Definition. (optional)

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

        # ListTaskDefinitions: List Task Definitions
        api_response = api_instance.list_task_definitions(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 TaskDefinitionsApi->list_task_definitions: %s\n" % e)

main()

Parameters

Name	Type	Description	Notes
as_at	datetime	The asAt datetime at which to list the Task Definitions. Defaults to return the latest version of each Task Definition 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 task definitions from a previous call to list task definitions. 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 whose values to return on each Task Definition.	[optional]

Return type

PagedResourceListOfTaskDefinition

HTTP request headers

Content-Type: Not defined
Accept: application/json

HTTP response details

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

list_tasks_for_task_definition

ResourceListOfTask list_tasks_for_task_definition(scope, code, as_at=as_at, property_keys=property_keys)

ListTasksForTaskDefinition: List Tasks for a Task Definition

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,
    TaskDefinitionsApi
)

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(TaskDefinitionsApi)
    scope = 'scope_example' # str | The scope that identifies a Task Definition
    code = 'code_example' # str | The code that identifies a Task Definition
    as_at = '2013-10-20T19:20:30+01:00' # datetime | The asAt datetime at which to list the Tasks. Defaults to return the latest version of each Task if not specified. (optional)
    property_keys = ['property_keys_example'] # List[str] | The property keys (in the TaskDefinition or Workflow domain) whose values to return on each Task. (optional)

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

        # ListTasksForTaskDefinition: List Tasks for a Task Definition
        api_response = api_instance.list_tasks_for_task_definition(scope, code, as_at=as_at, property_keys=property_keys)
        pprint(api_response)

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

main()

Parameters

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

Return type

ResourceListOfTask

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 tasks found for this Task Definition.	-
0	Error response	-

update_task_definition

TaskDefinition update_task_definition(scope, code, update_task_definition_request)

UpdateTaskDefinition: Update an existing Task Definition

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,
    TaskDefinitionsApi
)

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(TaskDefinitionsApi)
    scope = 'scope_example' # str | The scope that identifies a Task Definition
    code = 'code_example' # str | The code that identifies a Task Definition

    # 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_task_definition_request = UpdateTaskDefinitionRequest.from_json("")
    # update_task_definition_request = UpdateTaskDefinitionRequest.from_dict({})
    update_task_definition_request = UpdateTaskDefinitionRequest()

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

        # UpdateTaskDefinition: Update an existing Task Definition
        api_response = api_instance.update_task_definition(scope, code, update_task_definition_request)
        pprint(api_response)

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

main()

Parameters

Name	Type	Description
scope	str	The scope that identifies a Task Definition
code	str	The code that identifies a Task Definition
update_task_definition_request	UpdateTaskDefinitionRequest	The data to update a Task Definition

Return type

TaskDefinition

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	Task Definition not found.	-
0	Error response	-

upsert_task_definition_properties

BatchUpsertTaskDefinitionPropertiesResponse upsert_task_definition_properties(scope, code, request_body, success_mode=success_mode)

[EXPERIMENTAL] UpsertTaskDefinitionProperties: Add, update and remove properties on an existing Task Definition 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,
    TaskDefinitionsApi
)

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(TaskDefinitionsApi)
    scope = 'scope_example' # str | The scope that identifies a Task Definition
    code = 'code_example' # str | The code that identifies a Task Definition
    request_body = {"TaskDefinition/myScope/category":{"key":"TaskDefinition/myScope/category","value":{"labelValue":"exception-management"}},"TaskDefinition/myScope/priority":{"key":"TaskDefinition/myScope/priority","value":{"metricValue":{"value":1}}},"TaskDefinition/myScope/obsoleteProperty":{"key":"TaskDefinition/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_task_definition_properties(scope, code, request_body, success_mode=success_mode, opts=opts)

        # [EXPERIMENTAL] UpsertTaskDefinitionProperties: Add, update and remove properties on an existing Task Definition in bulk.
        api_response = api_instance.upsert_task_definition_properties(scope, code, request_body, success_mode=success_mode)
        pprint(api_response)

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

main()

Parameters

Name	Type	Description	Notes
scope	str	The scope that identifies a Task Definition
code	str	The code that identifies a Task Definition
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

BatchUpsertTaskDefinitionPropertiesResponse

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	Task Definition not found.	-
0	Error response	-

FilesExpand file tree

TaskDefinitionsApi.md

Latest commit

History

TaskDefinitionsApi.md

File metadata and controls

lusid_workflow.TaskDefinitionsApi

create_task_definition

Example

Parameters

Return type

HTTP request headers

HTTP response details

delete_task_definition

Example

Parameters

Return type

HTTP request headers

HTTP response details

get_task_definition

Example

Parameters

Return type

HTTP request headers

HTTP response details

list_task_definitions

Example

Parameters

Return type

HTTP request headers

HTTP response details

list_tasks_for_task_definition

Example

Parameters

Return type

HTTP request headers

HTTP response details

update_task_definition

Example

Parameters

Return type

HTTP request headers

HTTP response details

upsert_task_definition_properties

Example

Parameters

Return type

HTTP request headers

HTTP response details