Python: Functions
Invoke edge functions synchronously or asynchronously, deploy and version them, read logs and history, and manage access and bindings with the Python SDK.
result = await sovaria.fn_invoke("order-total", {"cart_id": cart_id})
print(result)Methods
sovaria.fn_delete
Fn delete.
await sovaria.fn_delete(function_name: str) -> dict[str, Any]Parameters
function_namestrrequired
- Returns
dict[str, Any]
sovaria.fn_deploy
Deploy a Forge function bundle (multipart upload).
await sovaria.fn_deploy(
bundle: bytes,
name: str,
*, project_id: str | None = None,
entrypoint: str = 'main.py',
handler: str = 'handler',
change_note: str = '',
) -> dict[str, Any]Parameters
bundlebytesrequirednamestrrequiredproject_idstr | NoneoptionaldefaultNoneentrypointstroptionaldefault'main.py'handlerstroptionaldefault'handler'change_notestroptionaldefault''
- Returns
dict[str, Any]
sovaria.fn_get
Fn get.
await sovaria.fn_get(
function_name: str,
*, project_id: str | None = None,
) -> dict[str, Any]Parameters
function_namestrrequiredproject_idstr | NoneoptionaldefaultNone
- Returns
dict[str, Any]
sovaria.fn_get_bindings
Fn get bindings.
await sovaria.fn_get_bindings(
function_name: str,
) -> list[dict[str, Any]]Parameters
function_namestrrequired
- Returns
list[dict[str, Any]]
sovaria.fn_get_policy
Fn get policy.
await sovaria.fn_get_policy(function_name: str) -> dict[str, Any]Parameters
function_namestrrequired
- Returns
dict[str, Any]
sovaria.fn_get_source
Fn get source.
await sovaria.fn_get_source(
function_name: str,
*, project_id: str | None = None,
) -> dict[str, Any]Parameters
function_namestrrequiredproject_idstr | NoneoptionaldefaultNone
- Returns
dict[str, Any]
sovaria.fn_invocation_history
List invocation history for a Forge function.
await sovaria.fn_invocation_history(
function_name: str,
*, project_id: str,
mode: str | None = None,
limit: int = 50,
offset: int = 0,
) -> dict[str, Any]Parameters
function_namestrrequiredproject_idstrrequiredmodestr | NoneoptionaldefaultNonelimitintoptionaldefault50offsetintoptionaldefault0
- Returns
dict[str, Any]
sovaria.fn_invocation_status
Get status for an async Forge invocation id.
await sovaria.fn_invocation_status(
invocation_id: str,
*, project_id: str,
) -> dict[str, Any]Parameters
invocation_idstrrequiredproject_idstrrequired
- Returns
dict[str, Any]
sovaria.fn_invoke
Invoke a Forge function inside project-scoped sandbox runtime.
await sovaria.fn_invoke(
function_name: str,
payload: dict[str, Any] | list[Any] | None = None,
*, project_id: str | None = None,
deploy_id: str | None = None,
execution_profile: str = 'standard',
) -> dict[str, Any]Parameters
function_namestrrequiredpayloaddict[str, Any] | list[Any] | NoneoptionaldefaultNoneproject_idstr | NoneoptionaldefaultNonedeploy_idstr | NoneoptionaldefaultNoneexecution_profilestroptionaldefault'standard'
- Returns
dict[str, Any]
sovaria.fn_invoke_async
Queue asynchronous Forge invocation for heavy execution profiles.
await sovaria.fn_invoke_async(
function_name: str,
payload: dict[str, Any] | list[Any] | None = None,
*, project_id: str,
deploy_id: str | None = None,
execution_profile: str = 'heavy',
simulation: bool = False,
) -> dict[str, Any]Parameters
function_namestrrequiredpayloaddict[str, Any] | list[Any] | NoneoptionaldefaultNoneproject_idstrrequireddeploy_idstr | NoneoptionaldefaultNoneexecution_profilestroptionaldefault'heavy'simulationbooloptionaldefaultFalse
- Returns
dict[str, Any]
sovaria.fn_jobs
List durable async jobs for a Forge function.
await sovaria.fn_jobs(
function_name: str,
*, project_id: str,
status: str | None = None,
limit: int = 100,
offset: int = 0,
) -> dict[str, Any]Parameters
function_namestrrequiredproject_idstrrequiredstatusstr | NoneoptionaldefaultNonelimitintoptionaldefault100offsetintoptionaldefault0
- Returns
dict[str, Any]
sovaria.fn_list
List Forge functions with active version and queue summary.
await sovaria.fn_list(
*, project_id: str,
limit: int = 100,
offset: int = 0,
) -> dict[str, Any]Parameters
project_idstrrequiredlimitintoptionaldefault100offsetintoptionaldefault0
- Returns
dict[str, Any]
sovaria.fn_logs
Fn logs.
await sovaria.fn_logs(
function_name: str,
*, limit: int = 50,
offset: int = 0,
status: str | None = None,
) -> list[dict[str, Any]]Parameters
function_namestrrequiredlimitintoptionaldefault50offsetintoptionaldefault0statusstr | NoneoptionaldefaultNone
- Returns
list[dict[str, Any]]
sovaria.fn_promote
Promote a deployment/version to active for a function.
await sovaria.fn_promote(
function_name: str,
*, project_id: str,
version: int | None = None,
deploy_id: str | None = None,
reason: str = '',
) -> dict[str, Any]Parameters
function_namestrrequiredproject_idstrrequiredversionint | NoneoptionaldefaultNonedeploy_idstr | NoneoptionaldefaultNonereasonstroptionaldefault''
- Returns
dict[str, Any]
sovaria.fn_request
Call a Forge endpoint under /api/v1/fn/*.
await sovaria.fn_request(
method: str,
path: str,
*, json: dict[str, Any] | list[Any] | None = None,
params: dict[str, str] | None = None,
requires_auth: bool = True,
) -> AnyParameters
methodstrrequiredpathstrrequiredjsondict[str, Any] | list[Any] | NoneoptionaldefaultNoneparamsdict[str, str] | NoneoptionaldefaultNonerequires_authbooloptionaldefaultTrue
- Returns
Any
sovaria.fn_rollback
Rollback active deployment to previous or explicit version.
await sovaria.fn_rollback(
function_name: str,
*, project_id: str,
version: int | None = None,
reason: str = '',
) -> dict[str, Any]Parameters
function_namestrrequiredproject_idstrrequiredversionint | NoneoptionaldefaultNonereasonstroptionaldefault''
- Returns
dict[str, Any]
sovaria.fn_simulate
Simulate a Forge function execution and persist simulation history.
await sovaria.fn_simulate(
function_name: str,
payload: dict[str, Any] | list[Any] | None = None,
*, project_id: str | None = None,
deploy_id: str | None = None,
execution_profile: str = 'standard',
) -> dict[str, Any]Parameters
function_namestrrequiredpayloaddict[str, Any] | list[Any] | NoneoptionaldefaultNoneproject_idstr | NoneoptionaldefaultNonedeploy_idstr | NoneoptionaldefaultNoneexecution_profilestroptionaldefault'standard'
- Returns
dict[str, Any]
sovaria.fn_simulation_diff
Diff two simulation invocation outputs.
await sovaria.fn_simulation_diff(
function_name: str,
*, project_id: str,
left_invocation_id: str,
right_invocation_id: str,
) -> dict[str, Any]Parameters
function_namestrrequiredproject_idstrrequiredleft_invocation_idstrrequiredright_invocation_idstrrequired
- Returns
dict[str, Any]
sovaria.fn_status
Get forge deployment status by deploy id.
await sovaria.fn_status(
deploy_id: str,
*, project_id: str | None = None,
) -> dict[str, Any]Parameters
deploy_idstrrequiredproject_idstr | NoneoptionaldefaultNone
- Returns
dict[str, Any]
sovaria.fn_update
Fn update.
await sovaria.fn_update(
function_name: str,
body: dict[str, Any],
*, project_id: str | None = None,
) -> dict[str, Any]Parameters
function_namestrrequiredbodydict[str, Any]requiredproject_idstr | NoneoptionaldefaultNone
- Returns
dict[str, Any]
sovaria.fn_update_bindings
Fn update bindings.
await sovaria.fn_update_bindings(
function_name: str,
bindings: list[dict[str, Any]],
) -> list[dict[str, Any]]Parameters
function_namestrrequiredbindingslist[dict[str, Any]]required
- Returns
list[dict[str, Any]]
sovaria.fn_update_policy
Fn update policy.
await sovaria.fn_update_policy(
function_name: str,
policy: dict[str, Any],
) -> dict[str, Any]Parameters
function_namestrrequiredpolicydict[str, Any]required
- Returns
dict[str, Any]
sovaria.fn_versions
List deployment versions for a Forge function.
await sovaria.fn_versions(
function_name: str,
*, project_id: str,
limit: int = 50,
offset: int = 0,
) -> dict[str, Any]Parameters
function_namestrrequiredproject_idstrrequiredlimitintoptionaldefault50offsetintoptionaldefault0
- Returns
dict[str, Any]