Invoke FunctionGraph Function from FunctionGraph using Temporary Credentials¶
This page demonstrates how to call a FunctionGraph implemented in Python from another FunctionGraph function using API calls and temporary security credentials (SecurityAccessKey/SecurityKey/SecurityToken) provided by an agency of Agency Type Cloud Service for Cloud Service FunctionGraph Service with permission to invoke FunctionGraph. for authentication.
Using temporary credentials the request has to be signed using the otc-api-sign-sdk-python.
See: Invoking FunctionGraph Event Function using API Calls for more details on how to use the REST API.
Prerequisites¶
URN of Function to be called. In this example the code of the function to be called is:
# -*- coding:utf-8 -*- import json def handler (event, context): return { "statusCode": 200, "isBase64Encoded": False, "body": json.dumps(event), "headers": { "Content-Type": "application/json" } }
Note
Ensure that the function and the subfunction are created in the same region.
An agency of Agency Type Cloud Service for Cloud Service FunctionGraph Service with permission to invoke FunctionGraph.
The permission policy should contain following policy statement:
{ "Version": "1.1", "Statement": [ { "Action": [ "functiongraph:function:invokeAsync*", "functiongraph:function:invoke" ], "Effect": "Allow" } ] }
or use an agency with default permission FunctionGraph CommonOperations.
Note
The permissions shown above are for demonstration purpose. Please follow the principle of least privilege when creating the permission policy for the agency.
e.g. to grant permission to invoke only specific functions, the policy statement should be like:
{ "Version": "1.1", "Statement": [ { "Action": [ "functiongraph:function:invokeAsync*", "functiongraph:function:invoke" ], "Effect": "Allow", "Resource": [ "RESOURCE_PATH" ] } ] }
where “RESOURCE_PATH” is in format
FunctionGraph:::function:group/function name
By adding Function name to the end of the generated prefix, you can define a specific path.
An asterisk * is allowed to indicate any function.
For example, FunctionGraph:*:*:function:default/* indicates any function in the default group.
For more details, see Policy Syntax in Identity and Access Management User Guide.
(Remark: changing the permission policy may take some time to take effect.)
Coding¶
index.py¶
Create a function with following content to call another FunctionGraph function:
import json
import os
from requests import request
from otc_api_sign_core import signer
def handler(event, context):
# Get temporary AK/SK/token from context.
ak = context.getSecurityAccessKey()
sk = context.getSecuritySecretKey()
token = context.getSecurityToken()
# Get the URN of the function to invoke from user data.
call_fg_urn = context.getUserData("CALL_FG_URN")
region = (call_fg_urn.split(":")[2] if call_fg_urn else "") or "eu-de"
fg_endpoint = f"https://functiongraph.{region}.otc.t-systems.com"
# Project ID is provided by FunctionGraph runtime environment.
project_id = os.environ.get("RUNTIME_PROJECT_ID", "")
# Synchronous invocation endpoint. For async, use .../invocations-async instead.
invoke_uri = f"{fg_endpoint}/v2/{project_id}/fgs/functions/{call_fg_urn}/invocations"
body = {
"key": "Hello FunctionGraph",
}
payload = json.dumps(body)
headers = {
"Content-Type": "application/json;charset=utf8",
"Host": f"functiongraph.{region}.otc.t-systems.com",
"X-Project-Id": project_id,
}
signed_request = signer.HttpRequest("POST", invoke_uri, headers, payload)
sig = signer.Signer()
sig.Key = ak
sig.Secret = sk
sig.SecurityToken = token
sig.Sign(signed_request)
response = request(
"POST",
invoke_uri,
headers=signed_request.headers,
data=payload.encode("utf-8"),
timeout=30,
)
response_body = response.text
print("Response:", response_body)
if response.status_code >= 400:
raise Exception(
f"Backend request failed with status {response.status_code}: {response_body}"
)
return response_body
Deployment¶
Create a deployment package using make create_package command and deploy the package to FunctionGraph using the console as event function from scratch using Python 3.10.
Configure the function:
set the handler name as src/index.handler.
specify an agency with permission to invoke FunctionGraph
and set the URN of the function to be called as Environment variable with key CALL_FG_URN.
Testing¶
Create a test event based on Blank Template and click Test.
Execution Result on the right should show a successful execution and the function set in the CALL_FG_URN environment variable should have a new invoke request in its Monitoring.