Functions

Functions API

Create function

When creating a function, the source code can be specified in one of three ways:

Via the folder argument, which is the path to the folder where the source code is located. function_path must point to a python file in the folder within which a function named handle must be defined.
Via the file_id argument, which is the ID of a zip-file uploaded to the files API. function_path must point to a python file in the zipped folder within which a function named handle must be defined.
Via the function_handle argument, which is a reference to a function object, which must be named handle.

The function named handle is the entrypoint of the created function. Valid arguments to handle are data, client, secrets and function_call_info: - If the user calls the function with input data, this is passed through the data argument. - If the user gives one or more secrets when creating the function, these are passed through the secrets argument. - Data about the function call can be accessed via the argument function_call_info, which is a dictionary with keys function_id, call_id, and, if the call is scheduled, schedule_id and scheduled_time.

By default, the function is deployed with the latest version of cognite-sdk. If a specific version is desired, it can be specified either in a requirements.txt file when deploying via the folder argument or between [requirements] tags when deploying via the function_handle argument (see example below).

For help with troubleshooting, please see this page.

Parameters

name (str) – The name of the function.
folder (str | None) – Path to the folder where the function source code is located.
file_id (int | None) – File ID of the code uploaded to the Files API.
function_path (str) – Relative path from the root folder to the file containing the handle function. Defaults to handler.py. Must be on POSIX path format.
function_handle (Callable | None) – Reference to a function object, which must be named handle.
external_id (str | None) – External id of the function.
description (str | None) – Description of the function.
owner (str | None) – Owner of this function. Typically used to know who created it.
secrets (dict | None) – Additional secrets as key/value pairs. These can e.g. password to simulators or other data sources. Keys must be lowercase characters, numbers or dashes (-) and at most 15 characters. You can create at most 30 secrets, all keys must be unique.
env_vars (dict | None) – Environment variables as key/value pairs. Keys can contain only letters, numbers or the underscore character. You can create at most 100 environment variables.
cpu (Number | None) – Number of CPU cores per function. Allowed values are in the range [0.1, 0.6], and None translates to the API default which is 0.25 in GCP. The argument is unavailable in Azure.
memory (Number | None) – Memory per function measured in GB. Allowed values are in the range [0.1, 2.5], and None translates to the API default which is 1 GB in GCP. The argument is unavailable in Azure.
runtime (str | None) – The function runtime. Valid values are [“py38”, “py39”, “py310”, “py311”, None], and None translates to the API default which will change over time. The runtime “py38” resolves to the latest version of the Python 3.8 series.
metadata (dict | None) – Metadata for the function as key/value pairs. Key & values can be at most 32, 512 characters long respectively. You can have at the most 16 key-value pairs, with a maximum size of 512 bytes.
index_url (str | None) – Index URL for Python Package Manager to use. Be aware of the intrinsic security implications of using the index_url option. More information can be found on official docs,
extra_index_urls (list[str] | None) –
Extra Index URLs for Python Package Manager to use. Be aware of the intrinsic security implications of using the extra_index_urls option. More information can be found on official docs,
skip_folder_validation (bool) – When creating a function using the ‘folder’ argument, pass True to skip the extra validation step that attempts to import the module. Skipping can be useful when your function requires several heavy packages to already be installed locally. Defaults to False.
data_set_id (int | None) – Data set to upload the function code to. Note: Does not affect the function itself.

Returns

The created function.

Return type

Function

Examples

Create function with source code in folder:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> function = c.functions.create(
...     name="myfunction",
...     folder="path/to/code",
...     function_path="path/to/function.py")

Create function with file_id from already uploaded source code:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> function = c.functions.create(
...     name="myfunction", file_id=123, function_path="path/to/function.py")

Create function with predefined function object named handle:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> function = c.functions.create(name="myfunction", function_handle=handle)

Create function with predefined function object named handle with dependencies:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>>
>>> def handle(client, data):
>>>     """
>>>     [requirements]
>>>     numpy
>>>     [/requirements]
>>>     """
>>>     ...
>>>
>>> function = c.functions.create(name="myfunction", function_handle=handle)

Note

When using a predefined function object, you can list dependencies between the tags [requirements] and [/requirements] in the function’s docstring. The dependencies will be parsed and validated in accordance with requirement format specified in PEP 508.

Delete function

FunctionsAPI.delete(id: int | Sequence[int] | None = None, external_id: str | Sequence[str] | None = None) → None

Delete one or more functions.

Parameters

id (int | Sequence[int] | None) – Id or list of ids.
external_id (str | Sequence[str] | None) – External ID or list of external ids.

Example

Delete functions by id or external id:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> c.functions.delete(id=[1,2,3], external_id="function3")

List functions

List all functions.

Parameters

name (str | None) – The name of the function.
owner (str | None) – Owner of the function.
file_id (int | None) – The file ID of the zip-file used to create the function.
status (str | None) – Status of the function. Possible values: [“Queued”, “Deploying”, “Ready”, “Failed”].
external_id_prefix (str | None) – External ID prefix to filter on.
created_time (dict[str, int] | TimestampRange | None) – Range between two timestamps. Possible keys are min and max, with values given as time stamps in ms.
limit (int | None) – Maximum number of functions to return. Pass in -1, float(‘inf’) or None to list all.

Returns

List of functions

Return type

FunctionList

Example

List functions:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> functions_list = c.functions.list()

Retrieve function

FunctionsAPI.retrieve(id: int | None = None, external_id: str | None = None) → Function | None

Retrieve a single function by id.

Parameters

id (int | None) – ID
external_id (str | None) – External ID

Returns

Requested function or None if it does not exist.

Return type

Function | None

Examples

Get function by id:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> res = c.functions.retrieve(id=1)

Get function by external id:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> res = c.functions.retrieve(external_id="abc")

Retrieve multiple functions

FunctionsAPI.retrieve_multiple(ids: Sequence[int] | None = None, external_ids: Sequence[str] | None = None, ignore_unknown_ids: bool = False) → FunctionList | Function | None

Retrieve multiple functions by id.

Parameters

ids (Sequence[int] | None) – IDs
external_ids (Sequence[str] | None) – External IDs
ignore_unknown_ids (bool) – Ignore IDs and external IDs that are not found rather than throw an exception.

Returns

The requested functions.

Return type

FunctionList | Function | None

Examples

Get function by id:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> res = c.functions.retrieve_multiple(ids=[1, 2, 3])

Get functions by external id:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> res = c.functions.retrieve_multiple(external_ids=["func1", "func2"])

Call function

FunctionsAPI.call(id: int | None = None, external_id: str | None = None, data: dict | None = None, wait: bool = True) → FunctionCall

Call a function by its ID or external ID..

Parameters

id (int | None) – ID
external_id (str | None) – External ID
data (dict | None) – Input data to the function (JSON serializable). This data is passed deserialized into the function through one of the arguments called data. WARNING: Secrets or other confidential information should not be passed via this argument. There is a dedicated secrets argument in FunctionsAPI.create() for this purpose.’
wait (bool) – Wait until the function call is finished. Defaults to True.

Returns

A function call object.

Return type

FunctionCall

Examples

Call a function by id:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> call = c.functions.call(id=1)

Call a function directly on the Function object:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> func = c.functions.retrieve(id=1)
>>> call = func.call()

Function calls

List function calls

List all calls associated with a specific function id. Either function_id or function_external_id must be specified.

Parameters

function_id (int | None) – ID of the function on which the calls were made.
function_external_id (str | None) – External ID of the function on which the calls were made.
status (str | None) – Status of the call. Possible values [“Running”, “Failed”, “Completed”, “Timeout”].
schedule_id (int | None) – Schedule id from which the call belongs (if any).
start_time (dict[str, int] | None) – Start time of the call. Possible keys are min and max, with values given as time stamps in ms.
end_time (dict[str, int] | None) – End time of the call. Possible keys are min and max, with values given as time stamps in ms.
limit (int | None) – Maximum number of function calls to list. Pass in -1, float(‘inf’) or None to list all Function Calls.

Returns

List of function calls

Return type

FunctionCallList

Examples

List function calls:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> calls = c.functions.calls.list(function_id=1)

List function calls directly on a function object:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> func = c.functions.retrieve(id=1)
>>> calls = func.list_calls()

Retrieve function call

FunctionCallsAPI.retrieve(call_id: int, function_id: int | None = None, function_external_id: str | None = None) → FunctionCall | None

Retrieve a single function call by id.

Parameters

call_id (int) – ID of the call.
function_id (int | None) – ID of the function on which the call was made.
function_external_id (str | None) – External ID of the function on which the call was made.

Returns

Requested function call or None if either call ID or function identifier is not found.

Return type

FunctionCall | None

Examples

Retrieve single function call by id:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> call = c.functions.calls.retrieve(call_id=2, function_id=1)

Retrieve function call directly on a function object:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> func = c.functions.retrieve(id=1)
>>> call = func.retrieve_call(id=2)

Retrieve function call response

FunctionCallsAPI.get_response(call_id: int, function_id: int | None = None, function_external_id: str | None = None) → dict | None

Retrieve the response from a function call.

Parameters

call_id (int) – ID of the call.
function_id (int | None) – ID of the function on which the call was made.
function_external_id (str | None) – External ID of the function on which the call was made.

Returns

Response from the function call.

Return type

dict | None

Examples

Retrieve function call response by call ID:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> response = c.functions.calls.get_response(call_id=2, function_id=1)

Retrieve function call response directly on a call object:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> call = c.functions.calls.retrieve(call_id=2, function_id=1)
>>> response = call.get_response()

Retrieve function call logs

FunctionCallsAPI.get_logs(call_id: int, function_id: int | None = None, function_external_id: str | None = None) → FunctionCallLog

Retrieve logs for function call.

Parameters

call_id (int) – ID of the call.
function_id (int | None) – ID of the function on which the call was made.
function_external_id (str | None) – External ID of the function on which the call was made.

Returns

Log for the function call.

Return type

FunctionCallLog

Examples

Retrieve function call logs by call ID:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> logs = c.functions.calls.get_logs(call_id=2, function_id=1)

Retrieve function call logs directly on a call object:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> call = c.functions.calls.retrieve(call_id=2, function_id=1)
>>> logs = call.get_logs()

Function schedules

List function schedules

List all schedules associated with a specific project.

Parameters

name (str | None) – Name of the function schedule.
function_id (int | None) – ID of the function the schedules are linked to.
function_external_id (str | None) – External ID of the function the schedules are linked to.
created_time (dict[str, int] | TimestampRange | None) – Range between two timestamps. Possible keys are min and max, with values given as time stamps in ms.
cron_expression (str | None) – Cron expression.
limit (int | None) – Maximum number of schedules to list. Pass in -1, float(‘inf’) or None to list all.

Returns

List of function schedules

Return type

FunctionSchedulesList

Examples

List function schedules:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> schedules = c.functions.schedules.list()

List schedules directly on a function object to get only schedules associated with this particular function:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> func = c.functions.retrieve(id=1)
>>> schedules = func.list_schedules(limit=None)

Create function schedule

FunctionSchedulesAPI.create(name: str, cron_expression: str, function_id: int | None = None, function_external_id: str | None = None, client_credentials: dict | ClientCredentials | None = None, description: str = '', data: dict | None = None) → FunctionSchedule

Create a schedule associated with a specific project.

Parameters

name (str) – Name of the schedule.
cron_expression (str) – Cron expression.
function_id (int | None) – Id of the function. This is required if the schedule is created with client_credentials.
function_external_id (str | None) – External id of the function. NOTE: This is deprecated and will be removed in a future major version.
client_credentials (dict | ClientCredentials | None) – (optional, ClientCredentials, Dict): Instance of ClientCredentials or a dictionary containing client credentials: client_id client_secret
description (str) – Description of the schedule.
data (dict | None) – Data to be passed to the scheduled run.

Returns

Created function schedule.

Return type

FunctionSchedule

Warning

Do not pass secrets or other confidential information via the data argument. There is a dedicated secrets argument in FunctionsAPI.create() for this purpose.

Examples

Create a function schedule that runs using specified client credentials (recommended):

>>> import os
>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes import ClientCredentials
>>> c = CogniteClient()
>>> schedule = c.functions.schedules.create(
...     name="My schedule",
...     function_id=123,
...     cron_expression="*/5 * * * *",
...     client_credentials=ClientCredentials("my-client-id", os.environ["MY_CLIENT_SECRET"]),
...     description="This schedule does magic stuff.",
...     data={"magic": "stuff"},
... )

You may also create a schedule that runs with your -current- credentials, i.e. the same credentials you used to instantiate the CogniteClient (that you’re using right now). Note: Unless you happen to already use client credentials, this is not a recommended way to create schedules, as it will create an explicit dependency on your user account, which it will run the function “on behalf of” (until the schedule is eventually removed):

>>> schedule = c.functions.schedules.create(
...     name="My schedule",
...     function_id=456,
...     cron_expression="*/5 * * * *",
...     description="A schedule just used for some temporary testing.",
... )

Delete function schedule

FunctionSchedulesAPI.delete(id: int) → None

Delete a schedule associated with a specific project.

Parameters: id (int) – Id of the schedule

Examples

Delete function schedule:

>>> from cognite.client import CogniteClient
>>> c = CogniteClient()
>>> c.functions.schedules.delete(id = 123)

Data classes

Bases: CogniteResource

A representation of a Cognite Function.

Parameters

id (int | None) – Id of the function.
name (str | None) – Name of the function.
external_id (str | None) – External id of the function.
description (str | None) – Description of the function.
owner (str | None) – Owner of the function.
status (str | None) – Status of the function.
file_id (int | None) – File id of the code represented by this object.
function_path (str | None) – Relative path from the root folder to the file containing the handle function. Defaults to handler.py. Must be on posix path format.
created_time (int | None) – Created time in UNIX.
secrets (dict | None) – Secrets attached to the function ((key, value) pairs).
env_vars (dict | None) – User specified environment variables on the function ((key, value) pairs).
cpu (Number | None) – Number of CPU cores per function. Defaults to 0.25. Allowed values are in the range [0.1, 0.6].
memory (Number | None) – Memory per function measured in GB. Defaults to 1. Allowed values are in the range [0.1, 2.5].
runtime (str | None) – Runtime of the function. Allowed values are [“py38”, “py39”,”py310”]. The runtime “py38” resolves to the latest version of the Python 3.8 series. Will default to “py38” if not specified.
runtime_version (str | None) – The complete specification of the function runtime with major, minor and patch version numbers.
metadata (dict | None) – Metadata associated with a function as a set of key:value pairs.
error (dict | None) – Dictionary with keys “message” and “trace”, which is populated if deployment fails.
cognite_client (CogniteClient | None) – An optional CogniteClient to associate with this data class.

call(data: dict | None = None, wait: bool = True) → FunctionCall

Call this particular function.

Parameters

data (dict | None) – Input data to the function (JSON serializable). This data is passed deserialized into the function through one of the arguments called data. WARNING: Secrets or other confidential information should not be passed via this argument. There is a dedicated secrets argument in FunctionsAPI.create() for this purpose.
wait (bool) – Wait until the function call is finished. Defaults to True.

Returns

A function call object.

Return type

FunctionCall

List all calls to this function.

Parameters

status (str | None) – Status of the call. Possible values [“Running”, “Failed”, “Completed”, “Timeout”].
schedule_id (int | None) – Schedule id from which the call belongs (if any).
start_time (dict[str, int] | None) – Start time of the call. Possible keys are min and max, with values given as time stamps in ms.
end_time (dict[str, int] | None) – End time of the call. Possible keys are min and max, with values given as time stamps in ms.
limit (int | None) – Maximum number of function calls to list. Pass in -1, float(‘inf’) or None to list all Function Calls.

Returns

List of function calls

Return type

FunctionCallList

list_schedules(limit: int | None = 25) → FunctionSchedulesList

List all schedules associated with this function.

Parameters: limit (int | None) – Maximum number of schedules to list. Pass in -1, float(‘inf’) or None to list all.
Returns: List of function schedules
Return type: FunctionSchedulesList

retrieve_call(id: int) → FunctionCall | None

Retrieve call by id.

Parameters: id (int) – ID of the call.
Returns: Requested function call or None if not found.
Return type: FunctionCall | None

update() → None: Update the function object. Can be useful to check for the latet status of the function (‘Queued’, ‘Deploying’, ‘Ready’ or ‘Failed’).

Bases: CogniteResource

A representation of a Cognite Function call.

Parameters

id (int | None) – A server-generated ID for the object.
start_time (int | None) – Start time of the call, measured in number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
end_time (int | None) – End time of the call, measured in number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
scheduled_time (int | None) – Scheduled time of the call, measured in number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
status (str | None) – Status of the function call (“Running”, “Completed” or “Failed”).
schedule_id (int | None) – The schedule id belonging to the call.
error (dict | None) – Error from the function call. It contains an error message and the stack trace.
function_id (int | None) – No description.
cognite_client (CogniteClient | None) – An optional CogniteClient to associate with this data class.

get_logs() → FunctionCallLog

Retrieve logs for this function call.

Returns: Log for the function call.
Return type: FunctionCallLog

get_response() → dict | None

Retrieve the response from this function call.

Returns: Response from the function call.
Return type: dict | None

update() → None: Update the function call object. Can be useful if the call was made with wait=False.

class cognite.client.data_classes.functions.FunctionCallList(resources: Collection[Any], cognite_client: CogniteClient | None = None): Bases: CogniteResourceList[FunctionCall]

class cognite.client.data_classes.functions.FunctionCallLog(resources: Collection[Any], cognite_client: CogniteClient | None = None)

Bases: CogniteResourceList[FunctionCallLogEntry]

A collection of function call log entries.

to_text(with_timestamps: bool = False) → str

Return a new-line delimited string of the log entry messages, optionally with entry timestamps.

Parameters: with_timestamps (bool) – Whether to include entry timestamps in the output. Defaults to False.
Returns: new-line delimited log entries.
Return type: str

class cognite.client.data_classes.functions.FunctionCallLogEntry(timestamp: int | None = None, message: str | None = None, cognite_client: CogniteClient | None = None)

Bases: CogniteResource

A log entry for a function call.

Parameters

timestamp (int | None) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
message (str | None) – Single line from stdout / stderr.
cognite_client (CogniteClient | None) – No description.

class cognite.client.data_classes.functions.FunctionList(resources: Collection[Any], cognite_client: CogniteClient | None = None): Bases: CogniteResourceList[Function], IdTransformerMixin

Bases: CogniteResource

A representation of a Cognite Function Schedule.

Parameters

id (int | None) – Id of the schedule.
name (str | None) – Name of the function schedule.
function_id (str | None) – Id of the function.
function_external_id (str | None) – External id of the function.
description (str | None) – Description of the function schedule.
created_time (int | None) – The number of milliseconds since 00:00:00 Thursday, 1 January 1970, Coordinated Universal Time (UTC), minus leap seconds.
cron_expression (str | None) – Cron expression
session_id (int | None) – ID of the session running with the schedule.
when (str | None) – When the schedule will trigger, in human readable text (server generated from cron_expression).
cognite_client (CogniteClient | None) – An optional CogniteClient to associate with this data class.

get_input_data() → dict | None

Retrieve the input data to the associated function.

Returns: Input data to the associated function or None if not set. This data is passed deserialized into the function through the data argument.
Return type: dict | None

class cognite.client.data_classes.functions.FunctionSchedulesList(resources: Collection[Any], cognite_client: CogniteClient | None = None): Bases: CogniteResourceList[FunctionSchedule]

class cognite.client.data_classes.functions.FunctionsLimits(timeout_minutes: int, cpu_cores: dict[str, float], memory_gb: dict[str, float], runtimes: list[str], response_size_mb: int | None = None)

Bases: CogniteResponse

Service limits for the associated project.

Parameters

timeout_minutes (int) – Timeout of each function call.
cpu_cores (dict[str, float]) – The number of CPU cores per function execution (i.e. function call).
memory_gb (dict[str, float]) – The amount of available memory in GB per function execution (i.e. function call).
runtimes (list[str]) – Available runtimes. For example, “py39” translates to the latest version of the Python 3.9.x series.
response_size_mb (int | None) – Maximum response size of function calls.

class cognite.client.data_classes.functions.FunctionsStatus(status: str)

Bases: CogniteResponse

Activation Status for the associated project.

Parameters: status (str) – Activation Status for the associated project.