FunctionGraph Service (FGS)

FunctionGraph hosts and computes event-driven functions in a serverless context while ensuring high availability, high scalability, and zero maintenance. All you need to do is write your code and set conditions.

Function

Function is a combination of code, runtime, resources, and settings required to achieve a specific purpose. It is the minimum unit that can run independently. A function can be triggered by triggers and automatically schedule required resources and environments to achieve expected results.

Function Lifecycle

Create Function

This API is used to create a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**attrs)

Delete Function

This API is used to delete a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**attrs)
conn.functiongraph.delete_function(fg)

List Functions

This API is used to query all functions.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

for fg in conn.functiongraph.functions():
    print(fg)

Get Function Code

This API is used to query the code of a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

code = conn.functiongraph.get_function_code(function='func_urn')

Get Function Metadata

This API is used to query the metadata of a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

meta = conn.functiongraph.get_function_metadata(function='func_urn')

Get Resource Tags

This API is used to query resource tags.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

ft = conn.functiongraph.get_resource_tags(function='func_urn')

Create Resource Tags

This API is used to create resource tags.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

conn.functiongraph.create_resource_tags(
    'func_urn',
    tags=[
        {
            'key': 'key',
            'value': 'value'
        },
        {
            'key': 'testKey2',
            'value': 'testValue2'
        }
    ]
)

Delete Resource Tags

This API is used to delete resource tags.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

conn.functiongraph.delete_resource_tags(
    'func_urn',
    tags=[
        {
            'key': 'testKey2',
            'value': 'testValue2'
        }
    ]
)

Update Function Code

This API is used to modify the code of a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

code_attrs = {
    'code_filename': 'index.zip',
    'code_type': 'inline',
    'func_code': {
        'file': 'UEsDBAoAAAAIAPQ1M1gNImPLrAAAAAEBAAAIAAAAaW5kZXgucHlN'
                'jtEOgjAMRd/5igVfxDAlxhjDo0S/wB+YrMgMdMvWGYnh390wEfrU'
                '3nvb0xXjG85qLRU+Sk8NP0UhUb3RltjTaUwkNKwVKDuwbA0vQMrD'
                'AhK8KSsTFsoCeYvsMw2xUkeCvKu0hLRk+6LIZ0u5s3BwPFwwUEEG'
                '/yo6B4vEXcshyBG+lb437kfNFpEWhATrQmqGTkYVH0Pit8FEdCqM'
                '6VQtSGncxYPpPz5O3fgFUEsBAh4DCgAAAAgA9DUzWA0iY8usAAAA'
                'AQEAAAgAAAAAAAAAAAAAAPMCAAAAAGluZGV4LnB5UEsFBgAAAAAB'
                'AAEANgAAANIAAAAAAA=='
    }
}
code = conn.functiongraph.update_function_code("func_urn", **code_attrs)

Update Function Metadata

This API is used to modify the metadata of a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

metadata_attrs = {
    'memory_size': 768,
    'handler': 'index.handler',
    'runtime': 'Python3.9',
    'mount_config': {
        'mount_user': {
            'user_id': -1,
            'user_group_id': -1
        },
    },
    'timeout': 40
}
metadata = conn.functiongraph.update_function_metadata(

Update Function Instances

This API is used to update the maximum number of instances of a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

metadata_attrs = {

Function Invocation

Function Execution Synchronously

This API is used to execute a function synchronously.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

func_urn = ('urn:fss:eu-de:45c274f200d2498683982c8741fb76ac:'
            'function:default:access-mysql-js-1213-1737554083545:latest')
inv = conn.functiongraph.executing_function_synchronously(
    func_urn, attrs={'a': 'b'}
)

Function Execution Asynchronously

This API is used to execute a function synchronously.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

func_urn = ('urn:fss:eu-de:45c274f200d2498683982c8741fb76ac:'
            'function:default:access-mysql-js-1213-1737554083545:latest')
inv = conn.functiongraph.executing_function_asynchronously(
    func_urn, attrs={'a': 'b'}
)

Function Quotas

Querying Tenant Quotas

This API is used to query tenant quotas.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

for q in conn.functiongraph.quotas():
    print(q)

Dependencies

Querying Dependencies

This API is used to query all dependencies.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

for d in conn.functiongraph.dependencies():
    print(d)

Creating a Dependency Version

This API is used to create a dependency version.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

u = 'https://fg-test-files.obs.eu-de.otc.t-systems.com/dependency.zip'
attrs = {
    'depend_link': u,
    'depend_type': 'obs',
    'runtime': 'Python3.10',
    'name': 'test-dep-1'
}
dv = conn.functiongraph.create_dependency_version(**attrs)
conn.functiongraph.delete_dependency_version(dv)

Querying Dependency Versions

This API is used to query dependency versions.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

u = 'https://fg-test-files.obs.eu-de.otc.t-systems.com/dependency.zip'
attrs = {
    'depend_link': u,
    'depend_type': 'obs',
    'runtime': 'Python3.10',
    'name': 'test-dep-1'
}
d = conn.functiongraph.create_dependency_version(**attrs)
for dv in conn.functiongraph.dependency_versions(d):
    print(dv)

Querying a Dependency Version

This API is used to query the details about a dependency version.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

u = 'https://fg-test-files.obs.eu-de.otc.t-systems.com/dependency.zip'
attrs = {
    'depend_link': u,
    'depend_type': 'obs',
    'runtime': 'Python3.10',
    'name': 'test-dep-1'
}
d = conn.functiongraph.create_dependency_version(**attrs)
dep = conn.functiongraph.get_dependency_version(d)

Deleting a Dependency Version

This API is used to delete a dependency version.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

u = 'https://fg-test-files.obs.eu-de.otc.t-systems.com/dependency.zip'
attrs = {
    'depend_link': u,
    'depend_type': 'obs',
    'runtime': 'Python3.10',
    'name': 'test-dep-1'
}
dv = conn.functiongraph.create_dependency_version(**attrs)
conn.functiongraph.delete_dependency_version(dv)

Test Events

Querying Test Events of a Function

This API is used to query the test events of a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

for fg in conn.functiongraph.events(func_urn='urn'):
    print(fg)

Creating a Test Event

This API is used to query the test events of a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

event_attrs = {
    'name': 'event-xx',
    'content': 'eyJrIjoidiJ9'
}
func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)
ev = conn.functiongraph.create_event(
    fg, **event_attrs
)

Deleting a Test Event

This API is used to delete a test event.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

event_attrs = {
    'name': 'event-xx',
    'content': 'eyJrIjoidiJ9'
}
func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)
ev = conn.functiongraph.create_event(
    fg, **event_attrs
)
conn.functiongraph.delete_event(fg, ev)

Obtaining the Details of a Test Event

This API is used to query the details of a test event.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

event_attrs = {
    'name': 'event-xx',
    'content': 'eyJrIjoidiJ9'
}
func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)
ev = conn.functiongraph.create_event(
    fg, **event_attrs
)
e = conn.functiongraph.get_event(
    fg, ev)
print(e)

Updating a Test Event

This API is used to update a test event.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

event_attrs = {
    'name': 'event-xx',
    'content': 'eyJrIjoidiJ9'
}
func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)
ev = conn.functiongraph.create_event(
    fg, **event_attrs
)
event_attrs = {
    'content': 'ewogICAgImJvZHkiOiAiIiwKICAgICJy'
}
updated = conn.functiongraph.update_event(
    fg, ev, **event_attrs)
print(updated)

Publishing a Function Version

This API is used to publish a function version.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)

publish_attrs = {
    'version': 'new-version',
    'description': 'otce',
}

publish = conn.functiongraph.publish_version(
    fg, **publish_attrs
)

Querying the Versions of a Function

This API is used to query the versions of a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

for v in conn.functiongraph.versions(func_urn='urn'):
    print(v)

Querying All Aliases of a Function

This API is used to query the versions and aliases of a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

for a in conn.functiongraph.aliases(func_urn='urn'):
    print(a)

Creating an Alias for a Function Version

This API is used to create an alias for a function version.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)

alias_attrs = {
    'name': 'a1',
    'version': 'new-version'
}
alias = conn.functiongraph.create_alias(
    fg.func_urn, **alias_attrs
)

Deleting an Alias of a Function Version

This API is used to delete an alias of a function version.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)
alias_attrs = {
    'name': 'a1',
    'version': 'new-version'
}
al = conn.functiongraph.create_alias(
    fg.func_urn, **alias_attrs
)
conn.functiongraph.delete_aliasw(fg, al)

Modifying the Alias of a Function Version

This API is used to modify the alias of a function version.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)

alias_attrs = {
    'name': 'a1',
    'version': 'new-version'
}
alias = conn.functiongraph.create_alias(
    fg.func_urn, **alias_attrs
)

new_attrs = {
    'version': 'new-version',
    'description': 'new',
}
updated = conn.functiongraph.update_alias(
    fg, alias, **new_attrs)

Querying the Alias of a Function Version

This API is used to query the alias of a function version.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

alias_attrs = {
    'name': 'a1',
    'version': 'new-version'
}
func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)
al = conn.functiongraph.create_alias(
    fg, **alias_attrs
)
a = conn.functiongraph.get_alias(
    fg, al)
print(a)

Querying Metrics in a Specified Period

This API is used to query the alias of a function version.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)
for m in conn.functiongraph.function_metrics(
        fg, period='1596679200000,1696679200000'):
    print(m)

Querying Tenant-Level Function Statistics

This API is used to query the alias of a function version.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

for m in conn.functiongraph.metrics(filter='monitor_data'):
    print(m)

Querying the Log Group and Stream of a Function

This API is used to query the LTS log group and stream settings of a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)

log = conn.functiongraph.get_lts_log_settings(fg)
print(log)

Enabling Log Reporting to LTS

This API is used to enable log reporting to LTS.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

conn.functiongraph.enable_lts_log()

Querying a Specified Function Template

This API is used to query a specified function template.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

template = conn.functiongraph.get_template(
    template_id="41d5d9ca-cea3-4ba9-b866-e30c46f45f1f"
)
print(template)

Querying Reserved Instances of a Function

This API is used to query reserved instances of a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

for i in conn.functiongraph.reserved_instances_config():
    print(i)

Querying the Number of Reserved Instances

This API is used to query the number of instances reserved for a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

for i in conn.functiongraph.reserved_instances():
    print(i)

Changing the Number of Reserved Instances

This API is used to change the number of reserved instances.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)

attrs = {
    "count": 2
}
instances = conn.functiongraph.update_instances_number(
    fg,
    **attrs
)

Importing a Function

This API is used to import a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

attrs = {
    'func_name': 'test',
    'file_type': 'zip',
    'file_name': 'test.zip',
    'file_code': 'UEsDBBQAAAAIAPBbPFpOs3AMAgEAAEwCAAAGAAAAZGVwLnB5jVHB'
                 'asMwDL37K4R3iSGMMtglsNNW2HH0B4oXq9SjtY2shJbSf5/tpl5y'
                 'GFQX23qS3nuWPQZPDD/ROyGEwR3stTMHpAZHdNxC7x3jiVUnIAXT'
                 '+XbJ8QRfmiIC7xGsCwND6YHGaNaqlhVom3PwVoieD16beCNQYj6O'
                 'bGrP4wL50Ro0kNtqRch4IzfYox0nsJPtjGExboM8kAMNhDF4F7Fi'
                 '90QSdKnJHDKy5iG+e4Oyg5fVql3C396cE7CTH9kOTUI6uPxJuMra'
                 'cp0RFinFvRmOITZ3CZNiPPUYGNblsD6pjoDzr/4sawEk8hRrvjy3'
                 'D9p5/d/OOs9JNiKnxasHLSzJlfgFUEsBAhQDFAAAAAgA8Fs8Wk6z'
                 'cAwCAQAATAIAAAYAAAAAAAAAAAAAAKSBAAAAAGRlcC5weVBLBQYA'
                 'AAAAAQABADQAAAAmAQAAAAA='
}
func = conn.functiongraph.import_function(**attrs)
print(func)

Exporting a Function

This API is used to export a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

event_attrs = {
    'name': 'event-xx',
    'content': 'eyJrIjoidiJ9'
}
func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)
exported = conn.functiongraph.export_function(fg, type="code")
print(exported)

Deleting All Triggers of a Function

This API is used to delete all triggers of a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)

conn.functiongraph.delete_all_triggers(
    fg.func_urn,
)

Deleting a Trigger

This API is used to delete a trigger.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)
trigger_attrs = {
    "trigger_type_code": "TIMER",
    "trigger_status": "ACTIVE",
    "event_data": {
        "name": "Timer-l8v2",
        "schedule": "3m",
        "schedule_type": "Rate"
    }
}

trigger = conn.functiongraph.create_trigger(
    fg, **trigger_attrs
)
conn.functiongraph.delete_trigger(
    fg.func_urn,
    trigger_attrs["trigger_type_code"],
    trigger.trigger_id
)

Querying All Triggers of a Function

This API is used to query all triggers of a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

for a in conn.functiongraph.triggers(function_urn='urn'):
    print(a)

Creating a Trigger

This API is used to create a trigger.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)

trigger_attrs = {
    "trigger_type_code": "TIMER",
    "trigger_status": "ACTIVE",
    "event_data": {
        "name": "Timer-l8v2",
        "schedule": "3m",
        "schedule_type": "Rate"
    }
}

trigger = conn.functiongraph.create_trigger(
    fg, **trigger_attrs
)

Querying a Trigger

This API is used to query a specified trigger.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)
trigger_attrs = {
    "trigger_type_code": "TIMER",
    "trigger_status": "ACTIVE",
    "event_data": {
        "name": "Timer-l8v2",
        "schedule": "3m",
        "schedule_type": "Rate"
    }
}

trigger = conn.functiongraph.create_trigger(
    fg, **trigger_attrs
)
tr = conn.functiongraph.get_trigger(
    fg.func_urn,
    trigger_attrs["trigger_type_code"],
    trigger.trigger_id)
print(tr)

Updating a Trigger

This API is used to update a trigger.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)

trigger_attrs = {
    "trigger_type_code": "TIMER",
    "trigger_status": "ACTIVE",
    "event_data": {
        "name": "Timer-l8v2",
        "schedule": "3m",
        "schedule_type": "Rate"
    }
}

trigger = conn.functiongraph.create_trigger(
    fg, **trigger_attrs
)

update_attrs = {
    "trigger_status": "DISABLED",
}
updated = conn.functiongraph.update_trigger(
    fg.func_urn,
    trigger_attrs["trigger_type_code"],
    trigger.trigger_id,
    **update_attrs
)

Querying Asynchronous Execution Notification Settings of a Function Version

This API is used to query the asynchronous invocation setting of a function version.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

for fg in conn.functiongraph.async_notifications(function='urn'):
    print(fg)

Deleting Asynchronous Execution Notification Settings

This API is used to delete the asynchronous execution notification settings of a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)
enable = conn.functiongraph.configure_async_notification(
    fg,
    max_async_event_age_in_seconds=1000,
)
conn.functiongraph.delete_async_notification(fg)

Configuring Asynchronous Execution Notification

This API is used to configure asynchronous execution notification for a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)

enable = conn.functiongraph.configure_async_notification(
    fg,
    max_async_event_age_in_seconds=1000,
)

Querying Asynchronous Execution Notification Settings of All Versions

This API is used to query the asynchronous execution notification settings of a function’s all versions.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

inv = conn.functiongraph.executing_function_asynchronously(
    function="urn", attrs={'a': 'b'}
)

for request in conn.functiongraph.async_notification_requests(
        function='urn',
        request_id=inv.request_id
):
    print(request)

Querying Asynchronous Invocation Requests

This API is used to query the asynchronous invocation requests of a function.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

for fg in conn.functiongraph.async_invocation_requests(function='urn'):
    print(fg)

Stopping an Asynchronous Invocation Request

This API is used to stop asynchronous invocation of a function with N concurrent instances. When calling this API, set recursive to false and force to true. The API will also stop the function’s other concurrent requests and return “4208 function invocation canceled”.

import openstack
from otcextensions import sdk

openstack.enable_logging(True)
conn = openstack.connect(cloud='otc')
sdk.register_otc_extensions(conn)

func_attrs = {
    'func_name': 'test-function',
    'package': 'default',
    'runtime': 'Python3.9',
    'handler': 'index.handler',
    'timeout': 30,
    'memory_size': 128,
    'code_type': 'inline',
}
fg = conn.functiongraph.create_function(**func_attrs)
inv = conn.functiongraph.executing_function_asynchronously(
    function="urn", attrs={'a': 'b'}
)
conn.functiongraph.stop_async_invocation_request(
    fg, request_id=inv.request_id)

last updated: UTC - commit: