diff --git a/.github/workflows/validate-integration.yml b/.github/workflows/validate-integration.yml
index f469090d..513cb153 100644
--- a/.github/workflows/validate-integration.yml
+++ b/.github/workflows/validate-integration.yml
@@ -3,6 +3,7 @@ name: Validate Integration (Tooling)
on:
pull_request:
branches: [master, main]
+ workflow_dispatch:
jobs:
validate:
@@ -20,4 +21,4 @@ jobs:
- name: Validate
uses: autohive-ai/autohive-integrations-tooling@1.0.2
with:
- base_ref: origin/${{ github.base_ref }}
+ base_ref: origin/${{ github.base_ref || 'master' }}
diff --git a/README.md b/README.md
index 00451968..68cd0b5d 100644
--- a/README.md
+++ b/README.md
@@ -70,7 +70,7 @@ Supports basic HTTP authentication and Bearer token authentication via the SDK.
### Coda
-[coda](coda): Comprehensive Coda integration for managing documents, pages, tables, and rows. Supports full CRUD operations for docs (list, get, create, update, delete) and pages (list, get, create with HTML/Markdown content, update metadata, delete). Includes table and column discovery (list tables/columns, get table/column details) and complete row management (list with filtering/sorting, get, upsert with keyColumns, update, delete single/multiple). Features Bearer token authentication, pagination support, async processing (HTTP 202 responses), multiple value formats (simple/rich), and comprehensive error handling. Ideal for document automation, content management, and data synchronization workflows.
+[coda](coda): Comprehensive Coda integration for managing documents, pages, tables, and rows. Supports full CRUD operations for docs (list, get, create, update, delete) and pages (list, get, create with HTML/Markdown content, update metadata, delete). Includes table and column discovery (list tables/columns, get table/column details) and complete row management (list with filtering/sorting, get, upsert with keyColumns, update, delete single/multiple). Features Bearer token (API token) authentication, pagination support, async processing (HTTP 202 responses), multiple value formats (simple/rich), and comprehensive error handling. Ideal for document automation, content management, and data synchronization workflows.
### ElevenLabs
diff --git a/aws/actions/cloudtrail.py b/aws/actions/cloudtrail.py
index 0c7cfb46..f090ec0a 100644
--- a/aws/actions/cloudtrail.py
+++ b/aws/actions/cloudtrail.py
@@ -22,13 +22,9 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
for attr in inputs["lookup_attributes"]
]
if inputs.get("start_time"):
- kwargs["StartTime"] = datetime.fromisoformat(
- inputs["start_time"].replace("Z", "+00:00")
- )
+ kwargs["StartTime"] = datetime.fromisoformat(inputs["start_time"].replace("Z", "+00:00"))
if inputs.get("end_time"):
- kwargs["EndTime"] = datetime.fromisoformat(
- inputs["end_time"].replace("Z", "+00:00")
- )
+ kwargs["EndTime"] = datetime.fromisoformat(inputs["end_time"].replace("Z", "+00:00"))
if inputs.get("next_token"):
kwargs["NextToken"] = inputs["next_token"]
response = await run_sync(client.lookup_events, **kwargs)
@@ -68,9 +64,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
client = create_boto3_client(context, "cloudtrail")
kwargs = {"Name": inputs["trail_name"]}
response = await run_sync(client.get_trail_status, **kwargs)
- trail_status = {
- k: v for k, v in response.items() if k != "ResponseMetadata"
- }
+ trail_status = {k: v for k, v in response.items() if k != "ResponseMetadata"}
return success_result({"trail_status": trail_status})
except Exception as e:
return error_result(e)
@@ -89,9 +83,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
{
"trail_arn": response.get("TrailARN"),
"event_selectors": response.get("EventSelectors", []),
- "advanced_event_selectors": response.get(
- "AdvancedEventSelectors", []
- ),
+ "advanced_event_selectors": response.get("AdvancedEventSelectors", []),
}
)
except Exception as e:
diff --git a/aws/actions/cloudwatch.py b/aws/actions/cloudwatch.py
index 4416499f..6cf7c926 100644
--- a/aws/actions/cloudwatch.py
+++ b/aws/actions/cloudwatch.py
@@ -39,9 +39,7 @@ class GetMetricDataAction(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
client = create_boto3_client(context, "cloudwatch")
- start_time = datetime.fromisoformat(
- inputs["start_time"].replace("Z", "+00:00")
- )
+ start_time = datetime.fromisoformat(inputs["start_time"].replace("Z", "+00:00"))
end_time = datetime.fromisoformat(inputs["end_time"].replace("Z", "+00:00"))
kwargs = {
"MetricDataQueries": inputs["metric_data_queries"],
@@ -49,9 +47,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
"EndTime": end_time,
}
response = await run_sync(client.get_metric_data, **kwargs)
- return success_result(
- {"metric_data_results": response.get("MetricDataResults", [])}
- )
+ return success_result({"metric_data_results": response.get("MetricDataResults", [])})
except Exception as e:
return error_result(e)
@@ -99,13 +95,9 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
if inputs.get("history_item_type"):
kwargs["HistoryItemType"] = inputs["history_item_type"]
if inputs.get("start_date"):
- kwargs["StartDate"] = datetime.fromisoformat(
- inputs["start_date"].replace("Z", "+00:00")
- )
+ kwargs["StartDate"] = datetime.fromisoformat(inputs["start_date"].replace("Z", "+00:00"))
if inputs.get("end_date"):
- kwargs["EndDate"] = datetime.fromisoformat(
- inputs["end_date"].replace("Z", "+00:00")
- )
+ kwargs["EndDate"] = datetime.fromisoformat(inputs["end_date"].replace("Z", "+00:00"))
if inputs.get("next_token"):
kwargs["NextToken"] = inputs["next_token"]
response = await run_sync(client.describe_alarm_history, **kwargs)
diff --git a/aws/actions/security_hub.py b/aws/actions/security_hub.py
index 06d1b40d..33884d1a 100644
--- a/aws/actions/security_hub.py
+++ b/aws/actions/security_hub.py
@@ -84,18 +84,14 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
for i in range(0, len(finding_arns), 100):
batch = finding_arns[i : i + 100]
lookup_kwargs = {
- "Filters": {
- "Id": [{"Value": arn, "Comparison": "EQUALS"} for arn in batch]
- },
+ "Filters": {"Id": [{"Value": arn, "Comparison": "EQUALS"} for arn in batch]},
"MaxResults": len(batch),
}
lookup_response = await run_sync(client.get_findings, **lookup_kwargs)
findings.extend(lookup_response.get("Findings", []))
# Build FindingIdentifiers from the looked-up findings
- finding_identifiers = [
- {"Id": f["Id"], "ProductArn": f["ProductArn"]} for f in findings
- ]
+ finding_identifiers = [{"Id": f["Id"], "ProductArn": f["ProductArn"]} for f in findings]
if not finding_identifiers:
return success_result(
@@ -164,21 +160,15 @@ async def fetch_insight_result(insight):
"group_by_attribute": insight.get("GroupByAttribute"),
}
try:
- result_response = await run_sync(
- client.get_insight_results, InsightArn=insight["InsightArn"]
- )
+ result_response = await run_sync(client.get_insight_results, InsightArn=insight["InsightArn"])
insight_data["results"] = result_response.get("InsightResults", {})
except Exception as inner_e:
insight_data["results"] = None
insight_data["error"] = str(inner_e)
return insight_data
- enriched_insights = await asyncio.gather(
- *[fetch_insight_result(insight) for insight in insights]
- )
+ enriched_insights = await asyncio.gather(*[fetch_insight_result(insight) for insight in insights])
- return success_result(
- {"insights": enriched_insights, "next_token": response.get("NextToken")}
- )
+ return success_result({"insights": enriched_insights, "next_token": response.get("NextToken")})
except Exception as e:
return error_result(e)
diff --git a/aws/helpers.py b/aws/helpers.py
index 99a8a86f..6c93c6a9 100644
--- a/aws/helpers.py
+++ b/aws/helpers.py
@@ -13,9 +13,7 @@ def create_boto3_client(context: ExecutionContext, service_name: str):
access_key = credentials.get("aws_access_key_id")
secret_key = credentials.get("aws_secret_access_key")
if not access_key or not secret_key:
- raise ValueError(
- "AWS credentials are missing: aws_access_key_id and aws_secret_access_key are required"
- )
+ raise ValueError("AWS credentials are missing: aws_access_key_id and aws_secret_access_key are required")
return boto3.client(
service_name,
aws_access_key_id=access_key,
@@ -53,6 +51,4 @@ def error_result(e: Exception) -> ActionResult:
if hasattr(e, "response"):
error_code = e.response.get("Error", {}).get("Code", "")
error_msg = e.response.get("Error", {}).get("Message", error_msg)
- return ActionResult(
- data={"result": False, "error": error_msg, "error_code": error_code}
- )
+ return ActionResult(data={"result": False, "error": error_msg, "error_code": error_code})
diff --git a/aws/tests/test_aws.py b/aws/tests/test_aws.py
index 2fab759f..619f2459 100644
--- a/aws/tests/test_aws.py
+++ b/aws/tests/test_aws.py
@@ -44,14 +44,10 @@ async def test_get_findings():
async def test_get_finding_details():
"""Test retrieving details for a specific Security Hub finding."""
print("\n=== Testing get_finding_details ===")
- inputs = {
- "finding_arn": "arn:aws:securityhub:us-east-1:123456789012:finding/example"
- }
+ inputs = {"finding_arn": "arn:aws:securityhub:us-east-1:123456789012:finding/example"}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
- result = await integration.execute_action(
- "get_finding_details", inputs, context
- )
+ result = await integration.execute_action("get_finding_details", inputs, context)
print(f"Result: {result}")
return result
except Exception as e:
@@ -69,9 +65,7 @@ async def test_update_finding_workflow():
}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
- result = await integration.execute_action(
- "update_finding_workflow", inputs, context
- )
+ result = await integration.execute_action("update_finding_workflow", inputs, context)
print(f"Result: {result}")
return result
except Exception as e:
@@ -118,9 +112,7 @@ async def test_list_guardduty_findings():
inputs = {"detector_id": "YOUR_DETECTOR_ID", "max_results": 10}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
- result = await integration.execute_action(
- "list_guardduty_findings", inputs, context
- )
+ result = await integration.execute_action("list_guardduty_findings", inputs, context)
print(f"Result: {result}")
return result
except Exception as e:
@@ -134,9 +126,7 @@ async def test_get_guardduty_finding_details():
inputs = {"detector_id": "YOUR_DETECTOR_ID", "finding_ids": ["example-finding-id"]}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
- result = await integration.execute_action(
- "get_guardduty_finding_details", inputs, context
- )
+ result = await integration.execute_action("get_guardduty_finding_details", inputs, context)
print(f"Result: {result}")
return result
except Exception as e:
@@ -150,9 +140,7 @@ async def test_archive_findings():
inputs = {"detector_id": "YOUR_DETECTOR_ID", "finding_ids": ["example-finding-id"]}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
- result = await integration.execute_action(
- "archive_findings", inputs, context
- )
+ result = await integration.execute_action("archive_findings", inputs, context)
print(f"Result: {result}")
return result
except Exception as e:
@@ -198,9 +186,7 @@ async def test_get_metric_data():
}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
- result = await integration.execute_action(
- "get_metric_data", inputs, context
- )
+ result = await integration.execute_action("get_metric_data", inputs, context)
print(f"Result: {result}")
return result
except Exception as e:
@@ -214,9 +200,7 @@ async def test_describe_alarms():
inputs = {"max_records": 10}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
- result = await integration.execute_action(
- "describe_alarms", inputs, context
- )
+ result = await integration.execute_action("describe_alarms", inputs, context)
print(f"Result: {result}")
return result
except Exception as e:
@@ -230,9 +214,7 @@ async def test_get_alarm_history():
inputs = {"max_records": 10}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
- result = await integration.execute_action(
- "get_alarm_history", inputs, context
- )
+ result = await integration.execute_action("get_alarm_history", inputs, context)
print(f"Result: {result}")
return result
except Exception as e:
@@ -250,9 +232,7 @@ async def test_set_alarm_state():
}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
- result = await integration.execute_action(
- "set_alarm_state", inputs, context
- )
+ result = await integration.execute_action("set_alarm_state", inputs, context)
print(f"Result: {result}")
return result
except Exception as e:
@@ -271,9 +251,7 @@ async def test_describe_log_groups():
inputs = {"limit": 10}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
- result = await integration.execute_action(
- "describe_log_groups", inputs, context
- )
+ result = await integration.execute_action("describe_log_groups", inputs, context)
print(f"Result: {result}")
return result
except Exception as e:
@@ -287,9 +265,7 @@ async def test_filter_log_events():
inputs = {"log_group_name": "/aws/lambda/test-function", "limit": 10}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
- result = await integration.execute_action(
- "filter_log_events", inputs, context
- )
+ result = await integration.execute_action("filter_log_events", inputs, context)
print(f"Result: {result}")
return result
except Exception as e:
@@ -340,9 +316,7 @@ async def test_describe_trails():
inputs = {}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
- result = await integration.execute_action(
- "describe_trails", inputs, context
- )
+ result = await integration.execute_action("describe_trails", inputs, context)
print(f"Result: {result}")
return result
except Exception as e:
@@ -356,9 +330,7 @@ async def test_get_trail_status():
inputs = {"trail_name": "management-events"}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
- result = await integration.execute_action(
- "get_trail_status", inputs, context
- )
+ result = await integration.execute_action("get_trail_status", inputs, context)
print(f"Result: {result}")
return result
except Exception as e:
@@ -372,9 +344,7 @@ async def test_get_event_selectors():
inputs = {"trail_name": "management-events"}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
- result = await integration.execute_action(
- "get_event_selectors", inputs, context
- )
+ result = await integration.execute_action("get_event_selectors", inputs, context)
print(f"Result: {result}")
return result
except Exception as e:
diff --git a/canva/canva.py b/canva/canva.py
index 82a2e11a..34d0c649 100644
--- a/canva/canva.py
+++ b/canva/canva.py
@@ -1,8 +1,12 @@
from autohive_integrations_sdk import (
- Integration, ExecutionContext, ActionHandler, ActionResult,
- ConnectedAccountHandler, ConnectedAccountInfo
+ Integration,
+ ExecutionContext,
+ ActionHandler,
+ ActionResult,
+ ConnectedAccountHandler,
+ ConnectedAccountInfo,
)
-from typing import Dict, Any, List, Optional
+from typing import Dict, Any
import base64
# Create the integration using the config.json
@@ -11,21 +15,16 @@
# ---- Connected Account Handler ----
+
@canva.connected_account()
class CanvaConnectedAccountHandler(ConnectedAccountHandler):
async def get_account_info(self, context: ExecutionContext) -> ConnectedAccountInfo:
"""Fetch Canva user information"""
# Get user profile (returns display name)
- profile_response = await context.fetch(
- f"{service_endpoint}/v1/users/me/profile",
- method="GET"
- )
+ profile_response = await context.fetch(f"{service_endpoint}/v1/users/me/profile", method="GET")
# Get user details (returns user_id and team_id)
- user_response = await context.fetch(
- f"{service_endpoint}/v1/users/me",
- method="GET"
- )
+ user_response = await context.fetch(f"{service_endpoint}/v1/users/me", method="GET")
# Extract information from responses
display_name = profile_response.get("profile", {}).get("display_name")
@@ -46,55 +45,47 @@ async def get_account_info(self, context: ExecutionContext) -> ConnectedAccountI
first_name=first_name,
last_name=last_name,
user_id=user_id,
- organization=team_id
+ organization=team_id,
)
+
# ---- Action Handlers ----
# User Actions
+
@canva.action("get_user_capabilities")
class GetUserCapabilities(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
- response = await context.fetch(
- f"{service_endpoint}/v1/users/me/capabilities",
- method="GET"
- )
+ response = await context.fetch(f"{service_endpoint}/v1/users/me/capabilities", method="GET")
return ActionResult(
- data={
- "result": True,
- "capabilities": response.get("capabilities", [])
- },
- cost_usd=0.0
+ data={"result": True, "capabilities": response.get("capabilities", [])},
+ cost_usd=0.0,
)
except Exception as e:
- return ActionResult(
- data={
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"result": False, "error": str(e)}, cost_usd=0.0)
+
# Asset Actions
+
@canva.action("upload_asset")
class UploadAsset(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
# Get file object from inputs (handle both 'file' and 'files' array)
- file_obj = inputs.get('file')
- files_arr = inputs.get('files')
+ file_obj = inputs.get("file")
+ files_arr = inputs.get("files")
if not file_obj and isinstance(files_arr, list) and files_arr:
file_obj = files_arr[0]
if not file_obj:
raise ValueError("No file provided")
- file_name = file_obj.get('name', 'asset')
- file_content_base64 = file_obj.get('content', '')
+ file_name = file_obj.get("name", "asset")
+ file_content_base64 = file_obj.get("content", "")
if not file_content_base64:
raise ValueError("File content is empty")
@@ -104,12 +95,12 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Encode the asset name in base64 for the header (max 50 chars)
asset_name = file_name[:50] # Truncate to 50 chars max
- name_base64 = base64.b64encode(asset_name.encode('utf-8')).decode('utf-8')
+ name_base64 = base64.b64encode(asset_name.encode("utf-8")).decode("utf-8")
# Prepare headers
headers = {
"Content-Type": "application/octet-stream",
- "Asset-Upload-Metadata": f'{{"name_base64": "{name_base64}"}}'
+ "Asset-Upload-Metadata": f'{{"name_base64": "{name_base64}"}}',
}
# Make the upload request with binary data
@@ -117,7 +108,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
f"{service_endpoint}/v1/asset-uploads",
method="POST",
headers=headers,
- data=file_data
+ data=file_data,
)
result = {"result": True}
@@ -128,29 +119,18 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
if job_data.get("status"):
result["status"] = job_data["status"]
- return ActionResult(
- data=result,
- cost_usd=0.0
- )
+ return ActionResult(data=result, cost_usd=0.0)
except Exception as e:
- return ActionResult(
- data={
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"result": False, "error": str(e)}, cost_usd=0.0)
+
@canva.action("get_asset_upload_status")
class GetAssetUploadStatus(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
- job_id = inputs['job_id']
+ job_id = inputs["job_id"]
- response = await context.fetch(
- f"{service_endpoint}/v1/asset-uploads/{job_id}",
- method="GET"
- )
+ response = await context.fetch(f"{service_endpoint}/v1/asset-uploads/{job_id}", method="GET")
result = {"result": True}
job_data = response.get("job", {})
@@ -160,147 +140,93 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
if job_data.get("asset"):
result["asset"] = job_data["asset"]
- return ActionResult(
- data=result,
- cost_usd=0.0
- )
+ return ActionResult(data=result, cost_usd=0.0)
except Exception as e:
- return ActionResult(
- data={
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"result": False, "error": str(e)}, cost_usd=0.0)
+
@canva.action("get_asset")
class GetAsset(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
- asset_id = inputs['asset_id']
+ asset_id = inputs["asset_id"]
- response = await context.fetch(
- f"{service_endpoint}/v1/assets/{asset_id}",
- method="GET"
- )
+ response = await context.fetch(f"{service_endpoint}/v1/assets/{asset_id}", method="GET")
result = {"result": True}
if response.get("asset"):
result["asset"] = response["asset"]
- return ActionResult(
- data=result,
- cost_usd=0.0
- )
+ return ActionResult(data=result, cost_usd=0.0)
except Exception as e:
- return ActionResult(
- data={
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"result": False, "error": str(e)}, cost_usd=0.0)
+
@canva.action("update_asset")
class UpdateAsset(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
- asset_id = inputs['asset_id']
+ asset_id = inputs["asset_id"]
# Build update payload
update_data = {}
- if 'name' in inputs:
- update_data['name'] = inputs['name']
- if 'tags' in inputs:
- update_data['tags'] = inputs['tags']
+ if "name" in inputs:
+ update_data["name"] = inputs["name"]
+ if "tags" in inputs:
+ update_data["tags"] = inputs["tags"]
- response = await context.fetch(
+ await context.fetch(
f"{service_endpoint}/v1/assets/{asset_id}",
method="PATCH",
- json=update_data
+ json=update_data,
)
- return ActionResult(
- data={"result": True},
- cost_usd=0.0
- )
+ return ActionResult(data={"result": True}, cost_usd=0.0)
except Exception as e:
- return ActionResult(
- data={
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"result": False, "error": str(e)}, cost_usd=0.0)
+
@canva.action("delete_asset")
class DeleteAsset(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
- asset_id = inputs['asset_id']
+ asset_id = inputs["asset_id"]
- response = await context.fetch(
- f"{service_endpoint}/v1/assets/{asset_id}",
- method="DELETE"
- )
+ await context.fetch(f"{service_endpoint}/v1/assets/{asset_id}", method="DELETE")
- return ActionResult(
- data={"result": True},
- cost_usd=0.0
- )
+ return ActionResult(data={"result": True}, cost_usd=0.0)
except Exception as e:
- return ActionResult(
- data={
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"result": False, "error": str(e)}, cost_usd=0.0)
+
# Design Actions
+
@canva.action("create_design")
class CreateDesign(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
# Build design creation payload
- design_data = {
- "design_type": {
- "type": "preset",
- "name": inputs['preset_type']
- }
- }
+ design_data = {"design_type": {"type": "preset", "name": inputs["preset_type"]}}
# Add optional fields
- if 'title' in inputs:
- design_data['title'] = inputs['title']
- if 'asset_id' in inputs:
- design_data['asset_id'] = inputs['asset_id']
+ if "title" in inputs:
+ design_data["title"] = inputs["title"]
+ if "asset_id" in inputs:
+ design_data["asset_id"] = inputs["asset_id"]
- response = await context.fetch(
- f"{service_endpoint}/v1/designs",
- method="POST",
- json=design_data
- )
+ response = await context.fetch(f"{service_endpoint}/v1/designs", method="POST", json=design_data)
result = {"result": True}
if response.get("design"):
result["design"] = response["design"]
- return ActionResult(
- data=result,
- cost_usd=0.0
- )
+ return ActionResult(data=result, cost_usd=0.0)
except Exception as e:
- return ActionResult(
- data={
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"result": False, "error": str(e)}, cost_usd=0.0)
+
@canva.action("list_designs")
class ListDesigns(ActionHandler):
@@ -308,183 +234,136 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
# Build query parameters
params = {}
- if 'query' in inputs:
- params['query'] = inputs['query']
- if 'continuation' in inputs:
- params['continuation'] = inputs['continuation']
- if 'ownership' in inputs:
- params['ownership'] = inputs['ownership']
- if 'sort_by' in inputs:
- params['sort_by'] = inputs['sort_by']
+ if "query" in inputs:
+ params["query"] = inputs["query"]
+ if "continuation" in inputs:
+ params["continuation"] = inputs["continuation"]
+ if "ownership" in inputs:
+ params["ownership"] = inputs["ownership"]
+ if "sort_by" in inputs:
+ params["sort_by"] = inputs["sort_by"]
- response = await context.fetch(
- f"{service_endpoint}/v1/designs",
- method="GET",
- params=params
- )
+ response = await context.fetch(f"{service_endpoint}/v1/designs", method="GET", params=params)
# Wrap response to match output schema
- result = {
- "designs": response.get("items", []),
- "result": True
- }
+ result = {"designs": response.get("items", []), "result": True}
# Only include continuation if it exists
if response.get("continuation"):
result["continuation"] = response["continuation"]
- return ActionResult(
- data=result,
- cost_usd=0.0
- )
+ return ActionResult(data=result, cost_usd=0.0)
except Exception as e:
- return ActionResult(
- data={
- "designs": [],
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"designs": [], "result": False, "error": str(e)}, cost_usd=0.0)
+
@canva.action("get_design")
class GetDesign(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
- design_id = inputs['design_id']
+ design_id = inputs["design_id"]
- response = await context.fetch(
- f"{service_endpoint}/v1/designs/{design_id}",
- method="GET"
- )
+ response = await context.fetch(f"{service_endpoint}/v1/designs/{design_id}", method="GET")
result = {"result": True}
if response.get("design"):
result["design"] = response["design"]
- return ActionResult(
- data=result,
- cost_usd=0.0
- )
+ return ActionResult(data=result, cost_usd=0.0)
except Exception as e:
- return ActionResult(
- data={
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"result": False, "error": str(e)}, cost_usd=0.0)
+
@canva.action("export_design")
class ExportDesign(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
- design_id = inputs['design_id']
- export_format = inputs['format']
+ design_id = inputs["design_id"]
+ export_format = inputs["format"]
# Build export format object
- format_obj = {
- "type": export_format
- }
+ format_obj = {"type": export_format}
# Add format-specific parameters
- if export_format.lower() == 'pdf':
+ if export_format.lower() == "pdf":
# PDF-specific parameters
- if 'export_quality' in inputs and inputs['export_quality']:
- format_obj['export_quality'] = inputs['export_quality']
- if 'paper_size' in inputs and inputs['paper_size']:
- format_obj['size'] = inputs['paper_size']
- if 'pages' in inputs and inputs['pages']:
- format_obj['pages'] = inputs['pages']
-
- elif export_format.lower() in ['jpg', 'jpeg']:
+ if "export_quality" in inputs and inputs["export_quality"]:
+ format_obj["export_quality"] = inputs["export_quality"]
+ if "paper_size" in inputs and inputs["paper_size"]:
+ format_obj["size"] = inputs["paper_size"]
+ if "pages" in inputs and inputs["pages"]:
+ format_obj["pages"] = inputs["pages"]
+
+ elif export_format.lower() in ["jpg", "jpeg"]:
# JPG parameters
- if 'jpg_quality' in inputs and inputs['jpg_quality']:
+ if "jpg_quality" in inputs and inputs["jpg_quality"]:
try:
- quality_val = int(inputs['jpg_quality'])
- format_obj['quality'] = max(1, min(100, quality_val))
+ quality_val = int(inputs["jpg_quality"])
+ format_obj["quality"] = max(1, min(100, quality_val))
except (ValueError, TypeError):
- format_obj['quality'] = 85
+ format_obj["quality"] = 85
else:
- format_obj['quality'] = 85
- if 'export_quality' in inputs and inputs['export_quality']:
- format_obj['export_quality'] = inputs['export_quality']
- if 'width' in inputs and inputs['width']:
- format_obj['width'] = inputs['width']
- if 'height' in inputs and inputs['height']:
- format_obj['height'] = inputs['height']
- if 'pages' in inputs and inputs['pages']:
- format_obj['pages'] = inputs['pages']
-
- elif export_format.lower() == 'png':
+ format_obj["quality"] = 85
+ if "export_quality" in inputs and inputs["export_quality"]:
+ format_obj["export_quality"] = inputs["export_quality"]
+ if "width" in inputs and inputs["width"]:
+ format_obj["width"] = inputs["width"]
+ if "height" in inputs and inputs["height"]:
+ format_obj["height"] = inputs["height"]
+ if "pages" in inputs and inputs["pages"]:
+ format_obj["pages"] = inputs["pages"]
+
+ elif export_format.lower() == "png":
# PNG parameters
- if 'export_quality' in inputs and inputs['export_quality']:
- format_obj['export_quality'] = inputs['export_quality']
- if 'width' in inputs and inputs['width']:
- format_obj['width'] = inputs['width']
- if 'height' in inputs and inputs['height']:
- format_obj['height'] = inputs['height']
- if 'lossless' in inputs and inputs['lossless'] is not None:
- format_obj['lossless'] = inputs['lossless']
- if 'transparent_background' in inputs and inputs['transparent_background'] is not None:
- format_obj['transparent_background'] = inputs['transparent_background']
- if 'as_single_image' in inputs and inputs['as_single_image'] is not None:
- format_obj['as_single_image'] = inputs['as_single_image']
- if 'pages' in inputs and inputs['pages']:
- format_obj['pages'] = inputs['pages']
-
- elif export_format.lower() in ['mp4', 'gif']:
+ if "export_quality" in inputs and inputs["export_quality"]:
+ format_obj["export_quality"] = inputs["export_quality"]
+ if "width" in inputs and inputs["width"]:
+ format_obj["width"] = inputs["width"]
+ if "height" in inputs and inputs["height"]:
+ format_obj["height"] = inputs["height"]
+ if "lossless" in inputs and inputs["lossless"] is not None:
+ format_obj["lossless"] = inputs["lossless"]
+ if "transparent_background" in inputs and inputs["transparent_background"] is not None:
+ format_obj["transparent_background"] = inputs["transparent_background"]
+ if "as_single_image" in inputs and inputs["as_single_image"] is not None:
+ format_obj["as_single_image"] = inputs["as_single_image"]
+ if "pages" in inputs and inputs["pages"]:
+ format_obj["pages"] = inputs["pages"]
+
+ elif export_format.lower() in ["mp4", "gif"]:
# MP4/GIF parameters - use quality with orientation_resolution
- if 'image_quality' in inputs and inputs['image_quality']:
- format_obj['quality'] = inputs['image_quality']
+ if "image_quality" in inputs and inputs["image_quality"]:
+ format_obj["quality"] = inputs["image_quality"]
else:
- format_obj['quality'] = 'horizontal_1080p'
- if 'export_quality' in inputs and inputs['export_quality']:
- format_obj['export_quality'] = inputs['export_quality']
- if 'pages' in inputs and inputs['pages']:
- format_obj['pages'] = inputs['pages']
+ format_obj["quality"] = "horizontal_1080p"
+ if "export_quality" in inputs and inputs["export_quality"]:
+ format_obj["export_quality"] = inputs["export_quality"]
+ if "pages" in inputs and inputs["pages"]:
+ format_obj["pages"] = inputs["pages"]
# Build export payload
- export_data = {
- "design_id": design_id,
- "format": format_obj
- }
+ export_data = {"design_id": design_id, "format": format_obj}
- response = await context.fetch(
- f"{service_endpoint}/v1/exports",
- method="POST",
- json=export_data
- )
+ response = await context.fetch(f"{service_endpoint}/v1/exports", method="POST", json=export_data)
job_id = response.get("job", {}).get("id")
return ActionResult(
- data={
- "result": True,
- "job_id": job_id
- } if job_id else {"result": True},
- cost_usd=0.0
+ data={"result": True, "job_id": job_id} if job_id else {"result": True},
+ cost_usd=0.0,
)
except Exception as e:
- return ActionResult(
- data={
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"result": False, "error": str(e)}, cost_usd=0.0)
+
@canva.action("get_export_status")
class GetExportStatus(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
- export_id = inputs['export_id']
+ export_id = inputs["export_id"]
- response = await context.fetch(
- f"{service_endpoint}/v1/exports/{export_id}",
- method="GET"
- )
+ response = await context.fetch(f"{service_endpoint}/v1/exports/{export_id}", method="GET")
result = {"result": True}
job_data = response.get("job", {})
@@ -494,35 +373,27 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
if job_data.get("urls"):
result["urls"] = job_data["urls"]
- return ActionResult(
- data=result,
- cost_usd=0.0
- )
+ return ActionResult(data=result, cost_usd=0.0)
except Exception as e:
- return ActionResult(
- data={
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"result": False, "error": str(e)}, cost_usd=0.0)
+
@canva.action("import_design")
class ImportDesign(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
# Get file object from inputs (handle both 'file' and 'files' array)
- file_obj = inputs.get('file')
- files_arr = inputs.get('files')
+ file_obj = inputs.get("file")
+ files_arr = inputs.get("files")
if not file_obj and isinstance(files_arr, list) and files_arr:
file_obj = files_arr[0]
if not file_obj:
raise ValueError("No file provided")
- file_name = file_obj.get('name', 'design')
- file_content_base64 = file_obj.get('content', '')
- mime_type = file_obj.get('contentType', 'application/pdf')
+ file_name = file_obj.get("name", "design")
+ file_content_base64 = file_obj.get("content", "")
+ mime_type = file_obj.get("contentType", "application/pdf")
if not file_content_base64:
raise ValueError("File content is empty")
@@ -531,21 +402,18 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
file_data = base64.b64decode(file_content_base64)
# Use provided title or filename
- title = inputs.get('title', file_name)
+ title = inputs.get("title", file_name)
# Encode the title in base64 for the header
- title_base64 = base64.b64encode(title.encode('utf-8')).decode('utf-8')
+ title_base64 = base64.b64encode(title.encode("utf-8")).decode("utf-8")
# Build Import-Metadata header
- metadata = {
- "title_base64": title_base64,
- "mime_type": mime_type
- }
+ metadata = {"title_base64": title_base64, "mime_type": mime_type}
# Prepare headers
headers = {
"Content-Type": "application/octet-stream",
- "Import-Metadata": str(metadata).replace("'", '"') # Convert to JSON format
+ "Import-Metadata": str(metadata).replace("'", '"'), # Convert to JSON format
}
# Make the import request with binary data
@@ -553,7 +421,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
f"{service_endpoint}/v1/imports",
method="POST",
headers=headers,
- data=file_data
+ data=file_data,
)
result = {"result": True}
@@ -564,29 +432,18 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
if job_data.get("status"):
result["status"] = job_data["status"]
- return ActionResult(
- data=result,
- cost_usd=0.0
- )
+ return ActionResult(data=result, cost_usd=0.0)
except Exception as e:
- return ActionResult(
- data={
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"result": False, "error": str(e)}, cost_usd=0.0)
+
@canva.action("get_design_import_status")
class GetDesignImportStatus(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
- job_id = inputs['job_id']
+ job_id = inputs["job_id"]
- response = await context.fetch(
- f"{service_endpoint}/v1/imports/{job_id}",
- method="GET"
- )
+ response = await context.fetch(f"{service_endpoint}/v1/imports/{job_id}", method="GET")
result = {"result": True}
job_data = response.get("job", {})
@@ -596,38 +453,23 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
if job_data.get("designs"):
result["designs"] = job_data["designs"]
- return ActionResult(
- data=result,
- cost_usd=0.0
- )
+ return ActionResult(data=result, cost_usd=0.0)
except Exception as e:
- return ActionResult(
- data={
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"result": False, "error": str(e)}, cost_usd=0.0)
+
@canva.action("import_design_from_url")
class ImportDesignFromUrl(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
# Build request body for URL-based import
- import_data = {
- "url": inputs['url'],
- "title": inputs['title']
- }
+ import_data = {"url": inputs["url"], "title": inputs["title"]}
# Add optional MIME type
- if 'mime_type' in inputs:
- import_data['mime_type'] = inputs['mime_type']
+ if "mime_type" in inputs:
+ import_data["mime_type"] = inputs["mime_type"]
- response = await context.fetch(
- f"{service_endpoint}/v1/url-imports",
- method="POST",
- json=import_data
- )
+ response = await context.fetch(f"{service_endpoint}/v1/url-imports", method="POST", json=import_data)
result = {"result": True}
job_data = response.get("job", {})
@@ -637,29 +479,18 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
if job_data.get("status"):
result["status"] = job_data["status"]
- return ActionResult(
- data=result,
- cost_usd=0.0
- )
+ return ActionResult(data=result, cost_usd=0.0)
except Exception as e:
- return ActionResult(
- data={
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"result": False, "error": str(e)}, cost_usd=0.0)
+
@canva.action("get_url_import_status")
class GetUrlImportStatus(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
- job_id = inputs['job_id']
+ job_id = inputs["job_id"]
- response = await context.fetch(
- f"{service_endpoint}/v1/url-imports/{job_id}",
- method="GET"
- )
+ response = await context.fetch(f"{service_endpoint}/v1/url-imports/{job_id}", method="GET")
result = {"result": True}
job_data = response.get("job", {})
@@ -669,21 +500,14 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
if job_data.get("designs"):
result["designs"] = job_data["designs"]
- return ActionResult(
- data=result,
- cost_usd=0.0
- )
+ return ActionResult(data=result, cost_usd=0.0)
except Exception as e:
- return ActionResult(
- data={
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"result": False, "error": str(e)}, cost_usd=0.0)
+
# Folder Actions
+
@canva.action("create_folder")
class CreateFolder(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
@@ -691,154 +515,95 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Build folder creation payload
# If parent_folder_id not provided, default to "root"
folder_data = {
- "name": inputs['name'],
- "parent_folder_id": inputs.get('parent_folder_id', 'root')
+ "name": inputs["name"],
+ "parent_folder_id": inputs.get("parent_folder_id", "root"),
}
- response = await context.fetch(
- f"{service_endpoint}/v1/folders",
- method="POST",
- json=folder_data
- )
+ response = await context.fetch(f"{service_endpoint}/v1/folders", method="POST", json=folder_data)
- return ActionResult(
- data={
- "result": True,
- "folder": response.get("folder")
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"result": True, "folder": response.get("folder")}, cost_usd=0.0)
except Exception as e:
- return ActionResult(
- data={
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"result": False, "error": str(e)}, cost_usd=0.0)
+
@canva.action("get_folder")
class GetFolder(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
- folder_id = inputs['folder_id']
+ folder_id = inputs["folder_id"]
- response = await context.fetch(
- f"{service_endpoint}/v1/folders/{folder_id}",
- method="GET"
- )
+ response = await context.fetch(f"{service_endpoint}/v1/folders/{folder_id}", method="GET")
result = {"result": True}
if response.get("folder"):
result["folder"] = response["folder"]
- return ActionResult(
- data=result,
- cost_usd=0.0
- )
+ return ActionResult(data=result, cost_usd=0.0)
except Exception as e:
- return ActionResult(
- data={
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"result": False, "error": str(e)}, cost_usd=0.0)
+
@canva.action("list_folder_items")
class ListFolderItems(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
- folder_id = inputs['folder_id']
+ folder_id = inputs["folder_id"]
# Build query parameters
params = {}
- if 'continuation' in inputs:
- params['continuation'] = inputs['continuation']
+ if "continuation" in inputs:
+ params["continuation"] = inputs["continuation"]
response = await context.fetch(
f"{service_endpoint}/v1/folders/{folder_id}/items",
method="GET",
- params=params
+ params=params,
)
- result = {
- "items": response.get("items", []),
- "result": True
- }
+ result = {"items": response.get("items", []), "result": True}
# Only include continuation if it exists
if response.get("continuation"):
result["continuation"] = response["continuation"]
- return ActionResult(
- data=result,
- cost_usd=0.0
- )
+ return ActionResult(data=result, cost_usd=0.0)
except Exception as e:
- return ActionResult(
- data={
- "items": [],
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"items": [], "result": False, "error": str(e)}, cost_usd=0.0)
+
@canva.action("update_folder")
class UpdateFolder(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
- folder_id = inputs['folder_id']
+ folder_id = inputs["folder_id"]
# Build update payload
- update_data = {
- "name": inputs['name']
- }
+ update_data = {"name": inputs["name"]}
- response = await context.fetch(
+ await context.fetch(
f"{service_endpoint}/v1/folders/{folder_id}",
method="PATCH",
- json=update_data
+ json=update_data,
)
- return ActionResult(
- data={"result": True},
- cost_usd=0.0
- )
+ return ActionResult(data={"result": True}, cost_usd=0.0)
except Exception as e:
- return ActionResult(
- data={
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"result": False, "error": str(e)}, cost_usd=0.0)
+
@canva.action("delete_folder")
class DeleteFolder(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
- folder_id = inputs['folder_id']
+ folder_id = inputs["folder_id"]
- response = await context.fetch(
- f"{service_endpoint}/v1/folders/{folder_id}",
- method="DELETE"
- )
+ await context.fetch(f"{service_endpoint}/v1/folders/{folder_id}", method="DELETE")
- return ActionResult(
- data={"result": True},
- cost_usd=0.0
- )
+ return ActionResult(data={"result": True}, cost_usd=0.0)
except Exception as e:
- return ActionResult(
- data={
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
+ return ActionResult(data={"result": False, "error": str(e)}, cost_usd=0.0)
+
@canva.action("move_item_to_folder")
class MoveItemToFolder(ActionHandler):
@@ -846,26 +611,12 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
# Build move payload - flat structure per Canva API
move_data = {
- "to_folder_id": inputs['destination_folder_id'],
- "item_id": inputs['item_id']
+ "to_folder_id": inputs["destination_folder_id"],
+ "item_id": inputs["item_id"],
}
- response = await context.fetch(
- f"{service_endpoint}/v1/folders/move",
- method="POST",
- json=move_data
- )
+ await context.fetch(f"{service_endpoint}/v1/folders/move", method="POST", json=move_data)
- return ActionResult(
- data={"result": True},
- cost_usd=0.0
- )
+ return ActionResult(data={"result": True}, cost_usd=0.0)
except Exception as e:
- return ActionResult(
- data={
- "result": False,
- "error": str(e)
- },
- cost_usd=0.0
- )
-
+ return ActionResult(data={"result": False, "error": str(e)}, cost_usd=0.0)
diff --git a/canva/requirements.txt b/canva/requirements.txt
index 76ea1825..b56fee2e 100644
--- a/canva/requirements.txt
+++ b/canva/requirements.txt
@@ -1 +1 @@
-autohive-integrations-sdk
+autohive-integrations-sdk~=1.0.2
diff --git a/canva/tests/context.py b/canva/tests/context.py
index 7a5e1b70..4e97343e 100644
--- a/canva/tests/context.py
+++ b/canva/tests/context.py
@@ -1,7 +1,6 @@
# -*- coding: utf-8 -*-
import sys
import os
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../dependencies")))
-from canva import canva
\ No newline at end of file
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../dependencies")))
diff --git a/canva/tests/test_canva.py b/canva/tests/test_canva.py
index 9764eab3..cf0ff665 100644
--- a/canva/tests/test_canva.py
+++ b/canva/tests/test_canva.py
@@ -5,11 +5,7 @@
# Test configuration
# IMPORTANT: Replace with your actual Canva OAuth credentials
-TEST_AUTH = {
- "credentials": {
- "access_token": "your_access_token_here"
- }
-}
+TEST_AUTH = {"credentials": {"access_token": "your_access_token_here"}} # nosec B105
# Store IDs for dependent tests
test_asset_id = None
@@ -29,8 +25,8 @@ async def test_get_user_capabilities():
try:
result = await canva.execute_action("get_user_capabilities", inputs, context)
- if result.data.get('result'):
- capabilities = result.data.get('capabilities', [])
+ if result.data.get("result"):
+ capabilities = result.data.get("capabilities", [])
print(f"✓ Found {len(capabilities)} capabilities")
# Show key capabilities
@@ -54,17 +50,17 @@ async def test_create_design():
inputs = {
"preset_type": "presentation",
- "title": "Test Presentation from Integration"
+ "title": "Test Presentation from Integration",
}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
result = await canva.execute_action("create_design", inputs, context)
- if result.data.get('result'):
- design = result.data.get('design', {})
+ if result.data.get("result"):
+ design = result.data.get("design", {})
global test_design_id
- test_design_id = design.get('id')
+ test_design_id = design.get("id")
print(f"✓ Created design: {design.get('title')}")
print(f" ID: {test_design_id}")
@@ -84,22 +80,20 @@ async def test_list_designs():
"""Test listing user's designs."""
print("\n[TEST] Listing user's designs...")
- inputs = {
- "sort_by": "modified_descending"
- }
+ inputs = {"sort_by": "modified_descending"}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
result = await canva.execute_action("list_designs", inputs, context)
- if result.data.get('result'):
- designs = result.data.get('designs', [])
+ if result.data.get("result"):
+ designs = result.data.get("designs", [])
print(f"✓ Found {len(designs)} design(s)")
if designs:
global test_design_id
if not test_design_id:
- test_design_id = designs[0].get('id')
+ test_design_id = designs[0].get("id")
# Show first few designs
for i, design in enumerate(designs[:3]):
@@ -123,16 +117,14 @@ async def test_get_design():
print(f"\n[TEST] Getting design details for {test_design_id}...")
- inputs = {
- "design_id": test_design_id
- }
+ inputs = {"design_id": test_design_id}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
result = await canva.execute_action("get_design", inputs, context)
- if result.data.get('result'):
- design = result.data.get('design', {})
+ if result.data.get("result"):
+ design = result.data.get("design", {})
print(f"✓ Retrieved design: {design.get('title')}")
print(f" Created: {design.get('created_at')}")
print(f" Updated: {design.get('updated_at')}")
@@ -153,14 +145,13 @@ async def test_upload_asset():
print(" NOTE: This will create an asset in your Canva account")
# Simple 1x1 red pixel PNG
- import base64
png_base64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8DwHwAFBQIAX8jx0gAAAABJRU5ErkJggg=="
inputs = {
"file": {
"content": png_base64,
"name": "test_image.png",
- "contentType": "image/png"
+ "contentType": "image/png",
}
}
@@ -168,11 +159,11 @@ async def test_upload_asset():
try:
result = await canva.execute_action("upload_asset", inputs, context)
- if result.data.get('result'):
+ if result.data.get("result"):
global test_upload_job_id
- test_upload_job_id = result.data.get('job_id')
+ test_upload_job_id = result.data.get("job_id")
- print(f"✓ Upload initiated")
+ print("✓ Upload initiated")
print(f" Job ID: {test_upload_job_id}")
print(f" Status: {result.data.get('status')}")
@@ -194,22 +185,20 @@ async def test_get_asset_upload_status():
print(f"\n[TEST] Checking upload status for job {test_upload_job_id}...")
- inputs = {
- "job_id": test_upload_job_id
- }
+ inputs = {"job_id": test_upload_job_id}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
result = await canva.execute_action("get_asset_upload_status", inputs, context)
- if result.data.get('result'):
- status = result.data.get('status')
+ if result.data.get("result"):
+ status = result.data.get("status")
print(f"✓ Upload status: {status}")
- if status == 'success':
- asset = result.data.get('asset', {})
+ if status == "success":
+ asset = result.data.get("asset", {})
global test_asset_id
- test_asset_id = asset.get('id')
+ test_asset_id = asset.get("id")
print(f" Asset ID: {test_asset_id}")
print(f" Asset Name: {asset.get('name')}")
@@ -227,19 +216,16 @@ async def test_create_folder():
"""Test creating a folder."""
print("\n[TEST] Creating a test folder...")
- inputs = {
- "name": "Test Folder from Integration",
- "parent_folder_id": "root"
- }
+ inputs = {"name": "Test Folder from Integration", "parent_folder_id": "root"}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
result = await canva.execute_action("create_folder", inputs, context)
- if result.data.get('result'):
- folder = result.data.get('folder', {})
+ if result.data.get("result"):
+ folder = result.data.get("folder", {})
global test_folder_id
- test_folder_id = folder.get('id')
+ test_folder_id = folder.get("id")
print(f"✓ Created folder: {folder.get('name')}")
print(f" ID: {test_folder_id}")
@@ -262,16 +248,14 @@ async def test_list_folder_items():
print(f"\n[TEST] Listing items in folder {test_folder_id}...")
- inputs = {
- "folder_id": test_folder_id
- }
+ inputs = {"folder_id": test_folder_id}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
result = await canva.execute_action("list_folder_items", inputs, context)
- if result.data.get('result'):
- items = result.data.get('items', [])
+ if result.data.get("result"):
+ items = result.data.get("items", [])
print(f"✓ Found {len(items)} item(s)")
if items:
@@ -296,20 +280,17 @@ async def test_export_design():
print(f"\n[TEST] Exporting design {test_design_id} to PDF...")
- inputs = {
- "design_id": test_design_id,
- "format": "pdf"
- }
+ inputs = {"design_id": test_design_id, "format": "pdf"}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
result = await canva.execute_action("export_design", inputs, context)
- if result.data.get('result'):
+ if result.data.get("result"):
global test_export_job_id
- test_export_job_id = result.data.get('job_id')
+ test_export_job_id = result.data.get("job_id")
- print(f"✓ Export initiated")
+ print("✓ Export initiated")
print(f" Job ID: {test_export_job_id}")
return result
@@ -330,20 +311,18 @@ async def test_get_export_status():
print(f"\n[TEST] Checking export status for job {test_export_job_id}...")
- inputs = {
- "export_id": test_export_job_id
- }
+ inputs = {"export_id": test_export_job_id}
async with ExecutionContext(auth=TEST_AUTH) as context:
try:
result = await canva.execute_action("get_export_status", inputs, context)
- if result.data.get('result'):
- status = result.data.get('status')
+ if result.data.get("result"):
+ status = result.data.get("status")
print(f"✓ Export status: {status}")
- if status == 'success':
- urls = result.data.get('urls', [])
+ if status == "success":
+ urls = result.data.get("urls", [])
print(f" Download URLs: {len(urls)}")
if urls:
print(f" First URL: {urls[0][:60]}...")
diff --git a/coda/__init__.py b/coda/__init__.py
new file mode 100644
index 00000000..8b137891
--- /dev/null
+++ b/coda/__init__.py
@@ -0,0 +1 @@
+
diff --git a/coda/coda.py b/coda/coda.py
index 9df36616..f4621a34 100644
--- a/coda/coda.py
+++ b/coda/coda.py
@@ -1,7 +1,5 @@
-from autohive_integrations_sdk import (
- Integration, ExecutionContext, ActionHandler
-)
-from typing import Dict, Any, List, Optional
+from autohive_integrations_sdk import Integration, ExecutionContext, ActionHandler
+from typing import Dict, Any
# Create the integration using the config.json
coda = Integration.load()
@@ -12,6 +10,7 @@
# ---- Helper Functions ----
+
def get_auth_headers(context: ExecutionContext) -> Dict[str, str]:
"""
Build authentication headers for Coda API requests.
@@ -25,14 +24,12 @@ def get_auth_headers(context: ExecutionContext) -> Dict[str, str]:
credentials = context.auth.get("credentials", {})
api_token = credentials.get("api_token", "")
- return {
- "Authorization": f"Bearer {api_token}",
- "Content-Type": "application/json"
- }
+ return {"Authorization": f"Bearer {api_token}", "Content-Type": "application/json"}
# ---- Action Handlers ----
+
@coda.action("list_docs")
class ListDocsAction(ActionHandler):
"""
@@ -68,27 +65,15 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Make API request
url = f"{CODA_API_BASE_URL}/docs"
- response = await context.fetch(
- url,
- method="GET",
- headers=headers,
- params=params
- )
+ response = await context.fetch(url, method="GET", headers=headers, params=params)
# Extract docs from response
docs = response.get("items", [])
- return {
- "docs": docs,
- "result": True
- }
+ return {"docs": docs, "result": True}
except Exception as e:
- return {
- "docs": [],
- "result": False,
- "error": str(e)
- }
+ return {"docs": [], "result": False, "error": str(e)}
@coda.action("get_doc")
@@ -107,23 +92,12 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Make API request
url = f"{CODA_API_BASE_URL}/docs/{doc_id}"
- response = await context.fetch(
- url,
- method="GET",
- headers=headers
- )
+ response = await context.fetch(url, method="GET", headers=headers)
- return {
- "data": response,
- "result": True
- }
+ return {"data": response, "result": True}
except Exception as e:
- return {
- "data": {},
- "result": False,
- "error": str(e)
- }
+ return {"data": {}, "result": False, "error": str(e)}
@coda.action("create_doc")
@@ -137,9 +111,7 @@ class CreateDocAction(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
# Build request body
- body = {
- "title": inputs["title"]
- }
+ body = {"title": inputs["title"]}
# Add optional fields if provided
if "source_doc" in inputs and inputs["source_doc"]:
@@ -156,24 +128,12 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Make API request
url = f"{CODA_API_BASE_URL}/docs"
- response = await context.fetch(
- url,
- method="POST",
- headers=headers,
- json=body
- )
-
- return {
- "data": response,
- "result": True
- }
+ response = await context.fetch(url, method="POST", headers=headers, json=body)
+
+ return {"data": response, "result": True}
except Exception as e:
- return {
- "data": {},
- "result": False,
- "error": str(e)
- }
+ return {"data": {}, "result": False, "error": str(e)}
@coda.action("update_doc")
@@ -202,24 +162,12 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Make API request
url = f"{CODA_API_BASE_URL}/docs/{doc_id}"
- response = await context.fetch(
- url,
- method="PATCH",
- headers=headers,
- json=body
- )
-
- return {
- "data": response,
- "result": True
- }
+ response = await context.fetch(url, method="PATCH", headers=headers, json=body)
+
+ return {"data": response, "result": True}
except Exception as e:
- return {
- "data": {},
- "result": False,
- "error": str(e)
- }
+ return {"data": {}, "result": False, "error": str(e)}
@coda.action("delete_doc")
@@ -240,23 +188,12 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Make API request
url = f"{CODA_API_BASE_URL}/docs/{doc_id}"
- response = await context.fetch(
- url,
- method="DELETE",
- headers=headers
- )
+ response = await context.fetch(url, method="DELETE", headers=headers)
- return {
- "data": response,
- "result": True
- }
+ return {"data": response, "result": True}
except Exception as e:
- return {
- "data": {},
- "result": False,
- "error": str(e)
- }
+ return {"data": {}, "result": False, "error": str(e)}
@coda.action("list_pages")
@@ -285,21 +222,13 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Make API request
url = f"{CODA_API_BASE_URL}/docs/{doc_id}/pages"
- response = await context.fetch(
- url,
- method="GET",
- headers=headers,
- params=params
- )
+ response = await context.fetch(url, method="GET", headers=headers, params=params)
# Extract pages from response
pages = response.get("items", [])
next_page_token = response.get("nextPageToken")
- result = {
- "pages": pages,
- "result": True
- }
+ result = {"pages": pages, "result": True}
if next_page_token:
result["next_page_token"] = next_page_token
@@ -307,11 +236,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
return result
except Exception as e:
- return {
- "pages": [],
- "result": False,
- "error": str(e)
- }
+ return {"pages": [], "result": False, "error": str(e)}
@coda.action("get_page")
@@ -331,23 +256,12 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Make API request
url = f"{CODA_API_BASE_URL}/docs/{doc_id}/pages/{page_id_or_name}"
- response = await context.fetch(
- url,
- method="GET",
- headers=headers
- )
+ response = await context.fetch(url, method="GET", headers=headers)
- return {
- "data": response,
- "result": True
- }
+ return {"data": response, "result": True}
except Exception as e:
- return {
- "data": {},
- "result": False,
- "error": str(e)
- }
+ return {"data": {}, "result": False, "error": str(e)}
@coda.action("create_page")
@@ -365,9 +279,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
name = inputs["name"]
# Build request body
- body = {
- "name": name
- }
+ body = {"name": name}
# Add optional fields
if "subtitle" in inputs and inputs["subtitle"]:
@@ -389,8 +301,8 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
"type": "canvas",
"canvasContent": {
"format": content_format,
- "content": inputs["content"]
- }
+ "content": inputs["content"],
+ },
}
# Get auth headers
@@ -398,24 +310,12 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Make API request
url = f"{CODA_API_BASE_URL}/docs/{doc_id}/pages"
- response = await context.fetch(
- url,
- method="POST",
- headers=headers,
- json=body
- )
-
- return {
- "data": response,
- "result": True
- }
+ response = await context.fetch(url, method="POST", headers=headers, json=body)
+
+ return {"data": response, "result": True}
except Exception as e:
- return {
- "data": {},
- "result": False,
- "error": str(e)
- }
+ return {"data": {}, "result": False, "error": str(e)}
@coda.action("update_page")
@@ -453,24 +353,12 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Make API request
url = f"{CODA_API_BASE_URL}/docs/{doc_id}/pages/{page_id_or_name}"
- response = await context.fetch(
- url,
- method="PUT",
- headers=headers,
- json=body
- )
-
- return {
- "data": response,
- "result": True
- }
+ response = await context.fetch(url, method="PUT", headers=headers, json=body)
+
+ return {"data": response, "result": True}
except Exception as e:
- return {
- "data": {},
- "result": False,
- "error": str(e)
- }
+ return {"data": {}, "result": False, "error": str(e)}
@coda.action("delete_page")
@@ -492,23 +380,12 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Make API request
url = f"{CODA_API_BASE_URL}/docs/{doc_id}/pages/{page_id_or_name}"
- response = await context.fetch(
- url,
- method="DELETE",
- headers=headers
- )
+ response = await context.fetch(url, method="DELETE", headers=headers)
- return {
- "data": response,
- "result": True
- }
+ return {"data": response, "result": True}
except Exception as e:
- return {
- "data": {},
- "result": False,
- "error": str(e)
- }
+ return {"data": {}, "result": False, "error": str(e)}
@coda.action("list_tables")
@@ -543,21 +420,13 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Make API request
url = f"{CODA_API_BASE_URL}/docs/{doc_id}/tables"
- response = await context.fetch(
- url,
- method="GET",
- headers=headers,
- params=params
- )
+ response = await context.fetch(url, method="GET", headers=headers, params=params)
# Extract tables from response
tables = response.get("items", [])
next_page_token = response.get("nextPageToken")
- result = {
- "tables": tables,
- "result": True
- }
+ result = {"tables": tables, "result": True}
if next_page_token:
result["next_page_token"] = next_page_token
@@ -565,11 +434,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
return result
except Exception as e:
- return {
- "tables": [],
- "result": False,
- "error": str(e)
- }
+ return {"tables": [], "result": False, "error": str(e)}
@coda.action("get_table")
@@ -589,23 +454,12 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Make API request
url = f"{CODA_API_BASE_URL}/docs/{doc_id}/tables/{table_id_or_name}"
- response = await context.fetch(
- url,
- method="GET",
- headers=headers
- )
+ response = await context.fetch(url, method="GET", headers=headers)
- return {
- "data": response,
- "result": True
- }
+ return {"data": response, "result": True}
except Exception as e:
- return {
- "data": {},
- "result": False,
- "error": str(e)
- }
+ return {"data": {}, "result": False, "error": str(e)}
@coda.action("list_columns")
@@ -638,21 +492,13 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Make API request
url = f"{CODA_API_BASE_URL}/docs/{doc_id}/tables/{table_id_or_name}/columns"
- response = await context.fetch(
- url,
- method="GET",
- headers=headers,
- params=params
- )
+ response = await context.fetch(url, method="GET", headers=headers, params=params)
# Extract columns from response
columns = response.get("items", [])
next_page_token = response.get("nextPageToken")
- result = {
- "columns": columns,
- "result": True
- }
+ result = {"columns": columns, "result": True}
if next_page_token:
result["next_page_token"] = next_page_token
@@ -660,11 +506,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
return result
except Exception as e:
- return {
- "columns": [],
- "result": False,
- "error": str(e)
- }
+ return {"columns": [], "result": False, "error": str(e)}
@coda.action("get_column")
@@ -685,23 +527,12 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Make API request
url = f"{CODA_API_BASE_URL}/docs/{doc_id}/tables/{table_id_or_name}/columns/{column_id_or_name}"
- response = await context.fetch(
- url,
- method="GET",
- headers=headers
- )
+ response = await context.fetch(url, method="GET", headers=headers)
- return {
- "data": response,
- "result": True
- }
+ return {"data": response, "result": True}
except Exception as e:
- return {
- "data": {},
- "result": False,
- "error": str(e)
- }
+ return {"data": {}, "result": False, "error": str(e)}
@coda.action("list_rows")
@@ -746,21 +577,13 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Make API request
url = f"{CODA_API_BASE_URL}/docs/{doc_id}/tables/{table_id_or_name}/rows"
- response = await context.fetch(
- url,
- method="GET",
- headers=headers,
- params=params
- )
+ response = await context.fetch(url, method="GET", headers=headers, params=params)
# Extract rows from response
rows = response.get("items", [])
next_page_token = response.get("nextPageToken")
- result = {
- "rows": rows,
- "result": True
- }
+ result = {"rows": rows, "result": True}
if next_page_token:
result["next_page_token"] = next_page_token
@@ -768,11 +591,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
return result
except Exception as e:
- return {
- "rows": [],
- "result": False,
- "error": str(e)
- }
+ return {"rows": [], "result": False, "error": str(e)}
@coda.action("get_row")
@@ -802,24 +621,12 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Make API request
url = f"{CODA_API_BASE_URL}/docs/{doc_id}/tables/{table_id_or_name}/rows/{row_id_or_name}"
- response = await context.fetch(
- url,
- method="GET",
- headers=headers,
- params=params
- )
-
- return {
- "data": response,
- "result": True
- }
+ response = await context.fetch(url, method="GET", headers=headers, params=params)
+
+ return {"data": response, "result": True}
except Exception as e:
- return {
- "data": {},
- "result": False,
- "error": str(e)
- }
+ return {"data": {}, "result": False, "error": str(e)}
@coda.action("upsert_rows")
@@ -838,9 +645,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
rows = inputs["rows"]
# Build request body
- body = {
- "rows": rows
- }
+ body = {"rows": rows}
# Add optional keyColumns for upsert behavior
if "key_columns" in inputs and inputs["key_columns"]:
@@ -856,25 +661,12 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Make API request
url = f"{CODA_API_BASE_URL}/docs/{doc_id}/tables/{table_id_or_name}/rows"
- response = await context.fetch(
- url,
- method="POST",
- headers=headers,
- params=params,
- json=body
- )
-
- return {
- "data": response,
- "result": True
- }
+ response = await context.fetch(url, method="POST", headers=headers, params=params, json=body)
+
+ return {"data": response, "result": True}
except Exception as e:
- return {
- "data": {},
- "result": False,
- "error": str(e)
- }
+ return {"data": {}, "result": False, "error": str(e)}
@coda.action("update_row")
@@ -894,11 +686,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
cells = inputs["cells"]
# Build request body
- body = {
- "row": {
- "cells": cells
- }
- }
+ body = {"row": {"cells": cells}}
# Build query parameters
params = {}
@@ -910,25 +698,12 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Make API request
url = f"{CODA_API_BASE_URL}/docs/{doc_id}/tables/{table_id_or_name}/rows/{row_id_or_name}"
- response = await context.fetch(
- url,
- method="PUT",
- headers=headers,
- params=params,
- json=body
- )
-
- return {
- "data": response,
- "result": True
- }
+ response = await context.fetch(url, method="PUT", headers=headers, params=params, json=body)
+
+ return {"data": response, "result": True}
except Exception as e:
- return {
- "data": {},
- "result": False,
- "error": str(e)
- }
+ return {"data": {}, "result": False, "error": str(e)}
@coda.action("delete_row")
@@ -950,23 +725,12 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Make API request
url = f"{CODA_API_BASE_URL}/docs/{doc_id}/tables/{table_id_or_name}/rows/{row_id_or_name}"
- response = await context.fetch(
- url,
- method="DELETE",
- headers=headers
- )
+ response = await context.fetch(url, method="DELETE", headers=headers)
- return {
- "data": response,
- "result": True
- }
+ return {"data": response, "result": True}
except Exception as e:
- return {
- "data": {},
- "result": False,
- "error": str(e)
- }
+ return {"data": {}, "result": False, "error": str(e)}
@coda.action("delete_rows")
@@ -984,30 +748,16 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
row_ids = inputs["row_ids"]
# Build request body
- body = {
- "rowIds": row_ids
- }
+ body = {"rowIds": row_ids}
# Get auth headers
headers = get_auth_headers(context)
# Make API request
url = f"{CODA_API_BASE_URL}/docs/{doc_id}/tables/{table_id_or_name}/rows"
- response = await context.fetch(
- url,
- method="DELETE",
- headers=headers,
- json=body
- )
-
- return {
- "data": response,
- "result": True
- }
+ response = await context.fetch(url, method="DELETE", headers=headers, json=body)
+
+ return {"data": response, "result": True}
except Exception as e:
- return {
- "data": {},
- "result": False,
- "error": str(e)
- }
+ return {"data": {}, "result": False, "error": str(e)}
diff --git a/coda/requirements.txt b/coda/requirements.txt
index 76ea1825..b56fee2e 100644
--- a/coda/requirements.txt
+++ b/coda/requirements.txt
@@ -1 +1 @@
-autohive-integrations-sdk
+autohive-integrations-sdk~=1.0.2
diff --git a/coda/tests/context.py b/coda/tests/context.py
index b3f151b9..4e97343e 100644
--- a/coda/tests/context.py
+++ b/coda/tests/context.py
@@ -1,7 +1,6 @@
# -*- coding: utf-8 -*-
import sys
import os
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../dependencies")))
-from coda import coda
\ No newline at end of file
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../dependencies")))
diff --git a/coda/tests/test_coda.py b/coda/tests/test_coda.py
index cb7ddd45..cda2d451 100644
--- a/coda/tests/test_coda.py
+++ b/coda/tests/test_coda.py
@@ -3,13 +3,10 @@
from context import coda
from autohive_integrations_sdk import ExecutionContext
+
async def test_list_docs():
"""Test listing all accessible Coda docs"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
inputs = {}
@@ -34,18 +31,12 @@ async def test_list_docs():
print(f"Error testing list_docs: {e}")
return None
+
async def test_list_docs_with_filters():
"""Test listing docs with filters"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
- inputs = {
- "is_owner": True,
- "limit": 5
- }
+ inputs = {"is_owner": True, "limit": 5}
async with ExecutionContext(auth=auth) as context:
try:
@@ -65,18 +56,12 @@ async def test_list_docs_with_filters():
print(f"Error testing list_docs_with_filters: {e}")
return None
+
async def test_list_docs_with_query():
"""Test searching docs by query"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
- inputs = {
- "query": "test",
- "limit": 10
- }
+ inputs = {"query": "test", "limit": 10}
async with ExecutionContext(auth=auth) as context:
try:
@@ -95,13 +80,10 @@ async def test_list_docs_with_query():
print(f"Error testing list_docs_with_query: {e}")
return None
+
async def test_get_doc():
"""Test getting a specific doc by ID"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
# First, get a doc ID from list_docs
list_inputs = {"limit": 1}
@@ -125,7 +107,7 @@ async def test_get_doc():
print(f"ERROR: {result.get('error')}")
else:
doc = result.get("data", {})
- print(f"SUCCESS: Retrieved doc")
+ print("SUCCESS: Retrieved doc")
print(f"Name: {doc.get('name')}")
print(f"ID: {doc.get('id')}")
print(f"Owner: {doc.get('owner')}")
@@ -137,17 +119,12 @@ async def test_get_doc():
print(f"Error testing get_doc: {e}")
return None
+
async def test_create_doc():
"""Test creating a new Coda doc"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
- inputs = {
- "title": "Test Doc - Created by Integration"
- }
+ inputs = {"title": "Test Doc - Created by Integration"}
async with ExecutionContext(auth=auth) as context:
try:
@@ -159,7 +136,7 @@ async def test_create_doc():
print(f"ERROR: {result.get('error')}")
else:
doc = result.get("data", {})
- print(f"SUCCESS: Doc created (HTTP 202 - Processing)")
+ print("SUCCESS: Doc created (HTTP 202 - Processing)")
print(f"Name: {doc.get('name')}")
print(f"ID: {doc.get('id')}")
print(f"Request ID: {doc.get('requestId', 'N/A')}")
@@ -169,13 +146,10 @@ async def test_create_doc():
print(f"Error testing create_doc: {e}")
return None
+
async def test_create_doc_from_source():
"""Test creating a doc from a source doc (template)"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
# First, get a doc ID to use as source
list_inputs = {"limit": 1}
@@ -192,17 +166,14 @@ async def test_create_doc_from_source():
print("\nTest 6: Create Doc from Source Template")
print("=" * 60)
- inputs = {
- "title": "Test Doc - From Template",
- "source_doc": source_doc_id
- }
+ inputs = {"title": "Test Doc - From Template", "source_doc": source_doc_id}
result = await coda.execute_action("create_doc", inputs, context)
if not result.get("result"):
print(f"ERROR: {result.get('error')}")
else:
doc = result.get("data", {})
- print(f"SUCCESS: Doc created from source (HTTP 202 - Processing)")
+ print("SUCCESS: Doc created from source (HTTP 202 - Processing)")
print(f"Name: {doc.get('name')}")
print(f"ID: {doc.get('id')}")
print(f"Source Doc: {source_doc_id}")
@@ -212,17 +183,12 @@ async def test_create_doc_from_source():
print(f"Error testing create_doc_from_source: {e}")
return None
+
async def test_list_published_docs():
"""Test listing only published docs"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
- inputs = {
- "is_published": True
- }
+ inputs = {"is_published": True}
async with ExecutionContext(auth=auth) as context:
try:
@@ -241,17 +207,12 @@ async def test_list_published_docs():
print(f"Error testing list_published_docs: {e}")
return None
+
async def test_list_starred_docs():
"""Test listing only starred docs"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
- inputs = {
- "is_starred": True
- }
+ inputs = {"is_starred": True}
async with ExecutionContext(auth=auth) as context:
try:
@@ -270,13 +231,10 @@ async def test_list_starred_docs():
print(f"Error testing list_starred_docs: {e}")
return None
+
async def test_update_doc():
"""Test updating a doc's metadata"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
# First, get a doc ID from list_docs
list_inputs = {"limit": 1}
@@ -296,7 +254,7 @@ async def test_update_doc():
inputs = {
"doc_id": doc_id,
"title": "Updated Doc Title via API",
- "icon_name": "star"
+ "icon_name": "star",
}
result = await coda.execute_action("update_doc", inputs, context)
@@ -304,7 +262,7 @@ async def test_update_doc():
print(f"ERROR: {result.get('error')}")
else:
doc = result.get("data", {})
- print(f"SUCCESS: Doc updated")
+ print("SUCCESS: Doc updated")
print(f"ID: {doc.get('id', doc_id)}")
return result
@@ -312,21 +270,16 @@ async def test_update_doc():
print(f"Error testing update_doc: {e}")
return None
+
async def test_delete_doc():
"""Test deleting a doc"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
# First, create a doc to delete
async with ExecutionContext(auth=auth) as context:
try:
# Create a test doc
- create_result = await coda.execute_action("create_doc", {
- "title": "Test Doc - To Be Deleted"
- }, context)
+ create_result = await coda.execute_action("create_doc", {"title": "Test Doc - To Be Deleted"}, context)
if not create_result.get("result"):
print("\nTest 10: Delete Doc (SKIPPED - Could not create test doc)")
@@ -334,6 +287,7 @@ async def test_delete_doc():
# Wait for doc creation to process
import asyncio
+
await asyncio.sleep(3)
# Get the created doc ID
@@ -352,7 +306,7 @@ async def test_delete_doc():
if not result.get("result"):
print(f"ERROR: {result.get('error')}")
else:
- print(f"SUCCESS: Doc deleted (HTTP 202 - Processing)")
+ print("SUCCESS: Doc deleted (HTTP 202 - Processing)")
print(f"Deleted doc ID: {doc_id}")
return result
@@ -360,13 +314,10 @@ async def test_delete_doc():
print(f"Error testing delete_doc: {e}")
return None
+
async def test_list_pages():
"""Test listing pages in a doc"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
# First, get a doc ID from list_docs
list_inputs = {"limit": 1}
@@ -400,13 +351,10 @@ async def test_list_pages():
print(f"Error testing list_pages: {e}")
return None
+
async def test_get_page():
"""Test getting a specific page"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
# First, get a doc ID and page ID
list_inputs = {"limit": 1}
@@ -439,7 +387,7 @@ async def test_get_page():
print(f"ERROR: {result.get('error')}")
else:
page = result.get("data", {})
- print(f"SUCCESS: Retrieved page")
+ print("SUCCESS: Retrieved page")
print(f"Name: {page.get('name')}")
print(f"ID: {page.get('id')}")
print(f"Subtitle: {page.get('subtitle', 'None')}")
@@ -450,13 +398,10 @@ async def test_get_page():
print(f"Error testing get_page: {e}")
return None
+
async def test_create_page():
"""Test creating a new page in a doc"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
# First, get a doc ID
list_inputs = {"limit": 1}
@@ -477,7 +422,7 @@ async def test_create_page():
"doc_id": doc_id,
"name": "Test Page - Created by Integration",
"subtitle": "This is a test page",
- "icon_name": "rocket"
+ "icon_name": "rocket",
}
result = await coda.execute_action("create_page", inputs, context)
@@ -485,7 +430,7 @@ async def test_create_page():
print(f"ERROR: {result.get('error')}")
else:
page = result.get("data", {})
- print(f"SUCCESS: Page created (HTTP 202 - Processing)")
+ print("SUCCESS: Page created (HTTP 202 - Processing)")
print(f"ID: {page.get('id')}")
print(f"Request ID: {page.get('requestId', 'N/A')}")
@@ -494,13 +439,10 @@ async def test_create_page():
print(f"Error testing create_page: {e}")
return None
+
async def test_create_page_with_content():
"""Test creating a page with HTML content"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
# First, get a doc ID
list_inputs = {"limit": 1}
@@ -521,7 +463,7 @@ async def test_create_page_with_content():
"doc_id": doc_id,
"name": "Test Page with Content",
"content_format": "html",
- "content": "Test Content
This is a test page with HTML content.
"
+ "content": "Test Content
This is a test page with HTML content.
",
}
result = await coda.execute_action("create_page", inputs, context)
@@ -529,7 +471,7 @@ async def test_create_page_with_content():
print(f"ERROR: {result.get('error')}")
else:
page = result.get("data", {})
- print(f"SUCCESS: Page with content created (HTTP 202 - Processing)")
+ print("SUCCESS: Page with content created (HTTP 202 - Processing)")
print(f"ID: {page.get('id')}")
return result
@@ -537,13 +479,10 @@ async def test_create_page_with_content():
print(f"Error testing create_page_with_content: {e}")
return None
+
async def test_update_page():
"""Test updating a page's metadata"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
# First, get a doc ID and page ID
list_inputs = {"limit": 1}
@@ -573,7 +512,7 @@ async def test_update_page():
"doc_id": doc_id,
"page_id_or_name": page_id,
"name": "Updated Page Name",
- "subtitle": "Updated subtitle"
+ "subtitle": "Updated subtitle",
}
result = await coda.execute_action("update_page", inputs, context)
@@ -581,7 +520,7 @@ async def test_update_page():
print(f"ERROR: {result.get('error')}")
else:
page = result.get("data", {})
- print(f"SUCCESS: Page updated (HTTP 202 - Processing)")
+ print("SUCCESS: Page updated (HTTP 202 - Processing)")
print(f"ID: {page.get('id')}")
print(f"Request ID: {page.get('requestId', 'N/A')}")
@@ -590,13 +529,10 @@ async def test_update_page():
print(f"Error testing update_page: {e}")
return None
+
async def test_delete_page():
"""Test deleting a page"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
# First, create a test page to delete
list_inputs = {"limit": 1}
@@ -611,10 +547,9 @@ async def test_delete_page():
doc_id = list_result["docs"][0]["id"]
# Create a page to delete
- create_result = await coda.execute_action("create_page", {
- "doc_id": doc_id,
- "name": "Page to Delete"
- }, context)
+ create_result = await coda.execute_action(
+ "create_page", {"doc_id": doc_id, "name": "Page to Delete"}, context
+ )
if not create_result.get("result"):
print("\nTest 14: Delete Page (SKIPPED - Could not create test page)")
@@ -622,6 +557,7 @@ async def test_delete_page():
# Wait a moment for page creation to process
import asyncio
+
await asyncio.sleep(2)
# Get the created page ID from list
@@ -641,17 +577,14 @@ async def test_delete_page():
print("\nTest 14: Delete Page")
print("=" * 60)
- inputs = {
- "doc_id": doc_id,
- "page_id_or_name": page_id
- }
+ inputs = {"doc_id": doc_id, "page_id_or_name": page_id}
result = await coda.execute_action("delete_page", inputs, context)
if not result.get("result"):
print(f"ERROR: {result.get('error')}")
else:
page = result.get("data", {})
- print(f"SUCCESS: Page deleted (HTTP 202 - Processing)")
+ print("SUCCESS: Page deleted (HTTP 202 - Processing)")
print(f"ID: {page.get('id')}")
print(f"Request ID: {page.get('requestId', 'N/A')}")
@@ -660,13 +593,10 @@ async def test_delete_page():
print(f"Error testing delete_page: {e}")
return None
+
async def test_list_tables():
"""Test listing tables in a doc"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
# First, get a doc ID
list_inputs = {"limit": 1}
@@ -702,13 +632,10 @@ async def test_list_tables():
print(f"Error testing list_tables: {e}")
return None
+
async def test_get_table():
"""Test getting a specific table"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
# First, get a doc ID and table ID
list_inputs = {"limit": 1}
@@ -741,7 +668,7 @@ async def test_get_table():
print(f"ERROR: {result.get('error')}")
else:
table = result.get("data", {})
- print(f"SUCCESS: Retrieved table")
+ print("SUCCESS: Retrieved table")
print(f"Name: {table.get('name')}")
print(f"ID: {table.get('id')}")
print(f"Type: {table.get('type')}")
@@ -753,13 +680,10 @@ async def test_get_table():
print(f"Error testing get_table: {e}")
return None
+
async def test_list_columns():
"""Test listing columns in a table"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
# First, get a doc ID and table ID
list_inputs = {"limit": 1}
@@ -804,13 +728,10 @@ async def test_list_columns():
print(f"Error testing list_columns: {e}")
return None
+
async def test_get_column():
"""Test getting a specific column"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
# First, get a doc ID, table ID, and column ID
list_inputs = {"limit": 1}
@@ -834,11 +755,11 @@ async def test_get_column():
table_id = tables_result["tables"][0]["id"]
# Get columns
- columns_result = await coda.execute_action("list_columns", {
- "doc_id": doc_id,
- "table_id_or_name": table_id,
- "limit": 1
- }, context)
+ columns_result = await coda.execute_action(
+ "list_columns",
+ {"doc_id": doc_id, "table_id_or_name": table_id, "limit": 1},
+ context,
+ )
if not columns_result.get("result") or not columns_result.get("columns"):
print("\nTest 18: Get Column (SKIPPED - No columns available)")
@@ -852,7 +773,7 @@ async def test_get_column():
inputs = {
"doc_id": doc_id,
"table_id_or_name": table_id,
- "column_id_or_name": column_id
+ "column_id_or_name": column_id,
}
result = await coda.execute_action("get_column", inputs, context)
@@ -860,7 +781,7 @@ async def test_get_column():
print(f"ERROR: {result.get('error')}")
else:
column = result.get("data", {})
- print(f"SUCCESS: Retrieved column")
+ print("SUCCESS: Retrieved column")
print(f"Name: {column.get('name')}")
print(f"ID: {column.get('id')}")
print(f"Value Type: {column.get('valueType')}")
@@ -872,13 +793,10 @@ async def test_get_column():
print(f"Error testing get_column: {e}")
return None
+
async def test_list_rows():
"""Test listing rows in a table"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
# Get doc and table IDs
async with ExecutionContext(auth=auth) as context:
@@ -916,13 +834,10 @@ async def test_list_rows():
print(f"Error testing list_rows: {e}")
return None
+
async def test_get_row():
"""Test getting a specific row"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
async with ExecutionContext(auth=auth) as context:
try:
@@ -939,7 +854,11 @@ async def test_get_row():
return None
table_id = tables_result["tables"][0]["id"]
- rows_result = await coda.execute_action("list_rows", {"doc_id": doc_id, "table_id_or_name": table_id, "limit": 1}, context)
+ rows_result = await coda.execute_action(
+ "list_rows",
+ {"doc_id": doc_id, "table_id_or_name": table_id, "limit": 1},
+ context,
+ )
if not rows_result.get("result") or not rows_result.get("rows"):
print("\nTest 20: Get Row (SKIPPED - No rows available)")
@@ -950,14 +869,18 @@ async def test_get_row():
print("\nTest 20: Get Row")
print("=" * 60)
- inputs = {"doc_id": doc_id, "table_id_or_name": table_id, "row_id_or_name": row_id}
+ inputs = {
+ "doc_id": doc_id,
+ "table_id_or_name": table_id,
+ "row_id_or_name": row_id,
+ }
result = await coda.execute_action("get_row", inputs, context)
if not result.get("result"):
print(f"ERROR: {result.get('error')}")
else:
row = result.get("data", {})
- print(f"SUCCESS: Retrieved row")
+ print("SUCCESS: Retrieved row")
print(f"ID: {row.get('id')}")
print(f"Created: {row.get('createdAt')}")
@@ -966,13 +889,10 @@ async def test_get_row():
print(f"Error testing get_row: {e}")
return None
+
async def test_upsert_rows():
"""Test upserting rows into a table"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
async with ExecutionContext(auth=auth) as context:
try:
@@ -982,14 +902,22 @@ async def test_upsert_rows():
return None
doc_id = list_result["docs"][0]["id"]
- tables_result = await coda.execute_action("list_tables", {"doc_id": doc_id, "table_types": "table", "limit": 1}, context)
+ tables_result = await coda.execute_action(
+ "list_tables",
+ {"doc_id": doc_id, "table_types": "table", "limit": 1},
+ context,
+ )
if not tables_result.get("result") or not tables_result.get("tables"):
print("\nTest 21: Upsert Rows (SKIPPED - No base tables available)")
return None
table_id = tables_result["tables"][0]["id"]
- columns_result = await coda.execute_action("list_columns", {"doc_id": doc_id, "table_id_or_name": table_id, "limit": 2}, context)
+ columns_result = await coda.execute_action(
+ "list_columns",
+ {"doc_id": doc_id, "table_id_or_name": table_id, "limit": 2},
+ context,
+ )
if not columns_result.get("result") or not columns_result.get("columns"):
print("\nTest 21: Upsert Rows (SKIPPED - No columns available)")
@@ -1004,9 +932,7 @@ async def test_upsert_rows():
inputs = {
"doc_id": doc_id,
"table_id_or_name": table_id,
- "rows": [
- {"cells": [{"column": column_id, "value": "Test Row from API"}]}
- ]
+ "rows": [{"cells": [{"column": column_id, "value": "Test Row from API"}]}],
}
result = await coda.execute_action("upsert_rows", inputs, context)
@@ -1014,7 +940,7 @@ async def test_upsert_rows():
print(f"ERROR: {result.get('error')}")
else:
data = result.get("data", {})
- print(f"SUCCESS: Rows upserted (HTTP 202 - Processing)")
+ print("SUCCESS: Rows upserted (HTTP 202 - Processing)")
print(f"Request ID: {data.get('requestId', 'N/A')}")
return result
@@ -1022,13 +948,10 @@ async def test_upsert_rows():
print(f"Error testing upsert_rows: {e}")
return None
+
async def test_update_row():
"""Test updating a row"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
async with ExecutionContext(auth=auth) as context:
try:
@@ -1045,14 +968,22 @@ async def test_update_row():
return None
table_id = tables_result["tables"][0]["id"]
- rows_result = await coda.execute_action("list_rows", {"doc_id": doc_id, "table_id_or_name": table_id, "limit": 1}, context)
+ rows_result = await coda.execute_action(
+ "list_rows",
+ {"doc_id": doc_id, "table_id_or_name": table_id, "limit": 1},
+ context,
+ )
if not rows_result.get("result") or not rows_result.get("rows"):
print("\nTest 22: Update Row (SKIPPED - No rows available)")
return None
row_id = rows_result["rows"][0]["id"]
- columns_result = await coda.execute_action("list_columns", {"doc_id": doc_id, "table_id_or_name": table_id, "limit": 1}, context)
+ columns_result = await coda.execute_action(
+ "list_columns",
+ {"doc_id": doc_id, "table_id_or_name": table_id, "limit": 1},
+ context,
+ )
if not columns_result.get("result") or not columns_result.get("columns"):
print("\nTest 22: Update Row (SKIPPED - No columns available)")
@@ -1067,27 +998,24 @@ async def test_update_row():
"doc_id": doc_id,
"table_id_or_name": table_id,
"row_id_or_name": row_id,
- "cells": [{"column": column_id, "value": "Updated via API"}]
+ "cells": [{"column": column_id, "value": "Updated via API"}],
}
result = await coda.execute_action("update_row", inputs, context)
if not result.get("result"):
print(f"ERROR: {result.get('error')}")
else:
- print(f"SUCCESS: Row updated (HTTP 202 - Processing)")
+ print("SUCCESS: Row updated (HTTP 202 - Processing)")
return result
except Exception as e:
print(f"Error testing update_row: {e}")
return None
+
async def test_delete_row():
"""Test deleting a single row"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
async with ExecutionContext(auth=auth) as context:
try:
@@ -1097,7 +1025,11 @@ async def test_delete_row():
return None
doc_id = list_result["docs"][0]["id"]
- tables_result = await coda.execute_action("list_tables", {"doc_id": doc_id, "table_types": "table", "limit": 1}, context)
+ tables_result = await coda.execute_action(
+ "list_tables",
+ {"doc_id": doc_id, "table_types": "table", "limit": 1},
+ context,
+ )
if not tables_result.get("result") or not tables_result.get("tables"):
print("\nTest 23: Delete Row (SKIPPED - No base tables available)")
@@ -1106,7 +1038,11 @@ async def test_delete_row():
table_id = tables_result["tables"][0]["id"]
# Create a test row to delete
- columns_result = await coda.execute_action("list_columns", {"doc_id": doc_id, "table_id_or_name": table_id, "limit": 1}, context)
+ columns_result = await coda.execute_action(
+ "list_columns",
+ {"doc_id": doc_id, "table_id_or_name": table_id, "limit": 1},
+ context,
+ )
if not columns_result.get("result") or not columns_result.get("columns"):
print("\nTest 23: Delete Row (SKIPPED - No columns available)")
return None
@@ -1114,11 +1050,15 @@ async def test_delete_row():
column_id = columns_result["columns"][0]["id"]
# Insert a test row
- upsert_result = await coda.execute_action("upsert_rows", {
- "doc_id": doc_id,
- "table_id_or_name": table_id,
- "rows": [{"cells": [{"column": column_id, "value": "Row to Delete"}]}]
- }, context)
+ upsert_result = await coda.execute_action(
+ "upsert_rows",
+ {
+ "doc_id": doc_id,
+ "table_id_or_name": table_id,
+ "rows": [{"cells": [{"column": column_id, "value": "Row to Delete"}]}],
+ },
+ context,
+ )
if not upsert_result.get("result"):
print("\nTest 23: Delete Row (SKIPPED - Could not create test row)")
@@ -1126,10 +1066,15 @@ async def test_delete_row():
# Wait for row creation
import asyncio
+
await asyncio.sleep(2)
# Get the row ID
- rows_result = await coda.execute_action("list_rows", {"doc_id": doc_id, "table_id_or_name": table_id, "limit": 10}, context)
+ rows_result = await coda.execute_action(
+ "list_rows",
+ {"doc_id": doc_id, "table_id_or_name": table_id, "limit": 10},
+ context,
+ )
row_id = None
for row in rows_result.get("rows", []):
values = row.get("values", {})
@@ -1147,27 +1092,24 @@ async def test_delete_row():
inputs = {
"doc_id": doc_id,
"table_id_or_name": table_id,
- "row_id_or_name": row_id
+ "row_id_or_name": row_id,
}
result = await coda.execute_action("delete_row", inputs, context)
if not result.get("result"):
print(f"ERROR: {result.get('error')}")
else:
- print(f"SUCCESS: Row deleted (HTTP 202 - Processing)")
+ print("SUCCESS: Row deleted (HTTP 202 - Processing)")
return result
except Exception as e:
print(f"Error testing delete_row: {e}")
return None
+
async def test_delete_rows():
"""Test deleting multiple rows"""
- auth = {
- "credentials": {
- "api_token": "your_api_token_here"
- }
- }
+ auth = {"credentials": {"api_token": "your_api_token_here"}} # nosec B105
async with ExecutionContext(auth=auth) as context:
try:
@@ -1177,7 +1119,11 @@ async def test_delete_rows():
return None
doc_id = list_result["docs"][0]["id"]
- tables_result = await coda.execute_action("list_tables", {"doc_id": doc_id, "table_types": "table", "limit": 1}, context)
+ tables_result = await coda.execute_action(
+ "list_tables",
+ {"doc_id": doc_id, "table_types": "table", "limit": 1},
+ context,
+ )
if not tables_result.get("result") or not tables_result.get("tables"):
print("\nTest 24: Delete Rows (SKIPPED - No base tables available)")
@@ -1186,7 +1132,11 @@ async def test_delete_rows():
table_id = tables_result["tables"][0]["id"]
# Create test rows to delete
- columns_result = await coda.execute_action("list_columns", {"doc_id": doc_id, "table_id_or_name": table_id, "limit": 1}, context)
+ columns_result = await coda.execute_action(
+ "list_columns",
+ {"doc_id": doc_id, "table_id_or_name": table_id, "limit": 1},
+ context,
+ )
if not columns_result.get("result") or not columns_result.get("columns"):
print("\nTest 24: Delete Rows (SKIPPED - No columns available)")
return None
@@ -1194,14 +1144,18 @@ async def test_delete_rows():
column_id = columns_result["columns"][0]["id"]
# Insert test rows
- upsert_result = await coda.execute_action("upsert_rows", {
- "doc_id": doc_id,
- "table_id_or_name": table_id,
- "rows": [
- {"cells": [{"column": column_id, "value": "Bulk Delete 1"}]},
- {"cells": [{"column": column_id, "value": "Bulk Delete 2"}]}
- ]
- }, context)
+ upsert_result = await coda.execute_action(
+ "upsert_rows",
+ {
+ "doc_id": doc_id,
+ "table_id_or_name": table_id,
+ "rows": [
+ {"cells": [{"column": column_id, "value": "Bulk Delete 1"}]},
+ {"cells": [{"column": column_id, "value": "Bulk Delete 2"}]},
+ ],
+ },
+ context,
+ )
if not upsert_result.get("result"):
print("\nTest 24: Delete Rows (SKIPPED - Could not create test rows)")
@@ -1209,10 +1163,13 @@ async def test_delete_rows():
# Wait for row creation
import asyncio
+
await asyncio.sleep(2)
# Get the row IDs
- rows_result = await coda.execute_action("list_rows", {"doc_id": doc_id, "table_id_or_name": table_id}, context)
+ rows_result = await coda.execute_action(
+ "list_rows", {"doc_id": doc_id, "table_id_or_name": table_id}, context
+ )
row_ids = []
for row in rows_result.get("rows", []):
values = row.get("values", {})
@@ -1231,7 +1188,7 @@ async def test_delete_rows():
inputs = {
"doc_id": doc_id,
"table_id_or_name": table_id,
- "row_ids": row_ids
+ "row_ids": row_ids,
}
result = await coda.execute_action("delete_rows", inputs, context)
@@ -1245,6 +1202,7 @@ async def test_delete_rows():
print(f"Error testing delete_rows: {e}")
return None
+
async def main():
"""Run all tests"""
print("\n" + "=" * 60)
@@ -1292,5 +1250,6 @@ async def main():
print("ALL TESTS COMPLETED")
print("=" * 60 + "\n")
+
if __name__ == "__main__":
asyncio.run(main())
diff --git a/companies-register/companies_register.py b/companies-register/companies_register.py
index 06339d4f..a45cc02a 100644
--- a/companies-register/companies_register.py
+++ b/companies-register/companies_register.py
@@ -1,5 +1,8 @@
from autohive_integrations_sdk import (
- Integration, ExecutionContext, ActionHandler, ActionResult
+ Integration,
+ ExecutionContext,
+ ActionHandler,
+ ActionResult,
)
from typing import Dict, Any
import aiohttp
@@ -19,7 +22,7 @@
BASE_URL_V2 = os.environ.get(
"COMPANIES_REGISTER_BASE_URL",
- "https://api.business.govt.nz/gateway/companies-office/companies-register/companies/v2"
+ "https://api.business.govt.nz/gateway/companies-office/companies-register/companies/v2",
)
# Sandbox: https://api.business.govt.nz/sandbox/companies-office/companies-register/companies/v2
@@ -29,30 +32,32 @@
# ---- Helper Functions ----
+
def safe_path(value: str) -> str:
"""URL-encode a path parameter to prevent path traversal."""
- return urllib.parse.quote(str(value), safe='')
+ return urllib.parse.quote(str(value), safe="")
def validate_request_id(request_id: str):
"""Validate requestId contains only safe characters."""
- if request_id and not re.match(r'^[a-zA-Z0-9\-]+$', request_id):
+ if request_id and not re.match(r"^[a-zA-Z0-9\-]+$", request_id):
raise ValueError("requestId must contain only alphanumeric characters and hyphens")
-
def get_api_headers(context: ExecutionContext, additional_headers: Dict[str, str] = None) -> Dict[str, str]:
"""Build headers for API requests."""
headers = {}
if not SUBSCRIPTION_KEY:
- raise ValueError("COMPANIES_REGISTER_SUBSCRIPTION_KEY is not set. Set this environment variable before making API calls.")
+ raise ValueError(
+ "COMPANIES_REGISTER_SUBSCRIPTION_KEY is not set. Set this environment variable before making API calls."
+ )
headers["Ocp-Apim-Subscription-Key"] = SUBSCRIPTION_KEY
- if hasattr(context, 'auth') and isinstance(context.auth, dict):
- credentials = context.auth.get('credentials', {})
+ if hasattr(context, "auth") and isinstance(context.auth, dict):
+ credentials = context.auth.get("credentials", {})
if isinstance(credentials, dict):
- access_token = credentials.get('access_token')
+ access_token = credentials.get("access_token")
if access_token:
headers["Authorization"] = f"Bearer {access_token}"
@@ -62,8 +67,13 @@ def get_api_headers(context: ExecutionContext, additional_headers: Dict[str, str
return headers
-async def fetch_with_headers(url: str, method: str = "GET", headers: Dict[str, str] = None,
- params: Dict[str, Any] = None, payload: Dict[str, Any] = None) -> tuple:
+async def fetch_with_headers(
+ url: str,
+ method: str = "GET",
+ headers: Dict[str, str] = None,
+ params: Dict[str, Any] = None,
+ payload: Dict[str, Any] = None,
+) -> tuple:
"""
Make an HTTP request using aiohttp and return both the response body and headers.
Needed because context.fetch() doesn't expose response headers (required for ETag).
@@ -71,12 +81,7 @@ async def fetch_with_headers(url: str, method: str = "GET", headers: Dict[str, s
async with aiohttp.ClientSession() as session:
request_headers = dict(headers) if headers else {}
- kwargs = {
- "method": method,
- "url": url,
- "headers": request_headers,
- "ssl": True
- }
+ kwargs = {"method": method, "url": url, "headers": request_headers, "ssl": True}
if params:
kwargs["params"] = params
@@ -87,13 +92,13 @@ async def fetch_with_headers(url: str, method: str = "GET", headers: Dict[str, s
request_headers["Content-Type"] = "application/json"
async with session.request(**kwargs) as response:
- etag = response.headers.get('ETag')
+ etag = response.headers.get("ETag")
response_headers = {}
for key, value in response.headers.items():
response_headers[key] = value
if etag:
- response_headers['ETag'] = etag
+ response_headers["ETag"] = etag
if response.status == 304:
return None, response_headers
@@ -112,6 +117,7 @@ async def fetch_with_headers(url: str, method: str = "GET", headers: Dict[str, s
# ---- Action Handlers ----
+
@companies_register.action("get_company_details")
class GetCompanyDetailsAction(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
@@ -137,11 +143,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
headers = get_api_headers(context, optional_headers)
- response, response_headers = await fetch_with_headers(
- url=url,
- method="GET",
- headers=headers
- )
+ response, response_headers = await fetch_with_headers(url=url, method="GET", headers=headers)
etag = response_headers.get("ETag")
@@ -150,9 +152,9 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
data={
"result": True,
"notModified": True,
- "message": "Company data not modified since last request"
+ "message": "Company data not modified since last request",
},
- cost_usd=None
+ cost_usd=None,
)
return ActionResult(
@@ -173,9 +175,9 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
"contacts": response.get("contacts"),
"link": response.get("link"),
"etag": etag,
- "result": True
+ "result": True,
},
- cost_usd=None
+ cost_usd=None,
)
except Exception as e:
@@ -186,18 +188,18 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
data={
"result": True,
"notModified": True,
- "message": "Company data not modified since last request"
+ "message": "Company data not modified since last request",
},
- cost_usd=None
+ cost_usd=None,
)
return ActionResult(
data={
"result": False,
"message": f"Error retrieving company details: {error_str}",
- "error": error_str
+ "error": error_str,
},
- cost_usd=None
+ cost_usd=None,
)
@@ -224,11 +226,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
headers = get_api_headers(context, optional_headers)
- response, response_headers = await fetch_with_headers(
- url=url,
- method="GET",
- headers=headers
- )
+ response, response_headers = await fetch_with_headers(url=url, method="GET", headers=headers)
etag = response_headers.get("ETag")
raw_contacts = response.get("contacts") if isinstance(response, dict) else None
@@ -241,9 +239,9 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
"phoneContacts": contacts.get("phoneContacts", []),
"emailAddresses": contacts.get("emailAddresses", []),
"etag": etag,
- "result": True
+ "result": True,
},
- cost_usd=None
+ cost_usd=None,
)
except Exception as e:
@@ -251,9 +249,9 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
data={
"result": False,
"message": f"Error retrieving company contacts: {str(e)}",
- "error": str(e)
+ "error": str(e),
},
- cost_usd=None
+ cost_usd=None,
)
@@ -308,9 +306,20 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
# Build payload with address lines only — exclude dpid
addr_payload = {}
- for field in ["addressId", "addressType", "addressPurpose", "careOf",
- "address1", "address2", "address3", "address4",
- "postCode", "countryCode", "description", "effectiveDate"]:
+ for field in [
+ "addressId",
+ "addressType",
+ "addressPurpose",
+ "careOf",
+ "address1",
+ "address2",
+ "address3",
+ "address4",
+ "postCode",
+ "countryCode",
+ "description",
+ "effectiveDate",
+ ]:
if address.get(field) is not None:
addr_payload[field] = address[field]
@@ -321,7 +330,13 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
if not phone.get("phoneNumber"):
raise ValueError("phoneNumber is required for phone contact updates")
phone_payload = {}
- for field in ["phoneContactId", "phoneNumber", "areaCode", "countryCode", "phonePurpose"]:
+ for field in [
+ "phoneContactId",
+ "phoneNumber",
+ "areaCode",
+ "countryCode",
+ "phonePurpose",
+ ]:
if phone.get(field) is not None:
phone_payload[field] = phone[field]
payload["phoneContact"] = phone_payload
@@ -339,19 +354,14 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
else:
raise ValueError(f"Invalid contactType '{contact_type}'. Use: address, phone, or email")
- optional_headers = {
- "If-Match": etag
- }
+ optional_headers = {"If-Match": etag}
if request_id:
optional_headers["api-business-govt-nz-Request-Id"] = request_id
headers = get_api_headers(context, optional_headers)
response, response_headers = await fetch_with_headers(
- url=url,
- method="PUT",
- headers=headers,
- payload=payload
+ url=url, method="PUT", headers=headers, payload=payload
)
new_etag = response_headers.get("ETag")
@@ -364,9 +374,9 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
"contact": contact,
"etag": new_etag,
"result": True,
- "message": "Contact updated successfully"
+ "message": "Contact updated successfully",
},
- cost_usd=None
+ cost_usd=None,
)
except Exception as e:
@@ -374,9 +384,9 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
data={
"result": False,
"message": f"Error updating company contact: {str(e)}",
- "error": str(e)
+ "error": str(e),
},
- cost_usd=None
+ cost_usd=None,
)
@@ -419,9 +429,20 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
)
addr_payload = {}
- for field in ["addressType", "addressPurpose", "dpid", "careOf",
- "address1", "address2", "address3", "address4",
- "postCode", "countryCode", "description", "effectiveDate"]:
+ for field in [
+ "addressType",
+ "addressPurpose",
+ "dpid",
+ "careOf",
+ "address1",
+ "address2",
+ "address3",
+ "address4",
+ "postCode",
+ "countryCode",
+ "description",
+ "effectiveDate",
+ ]:
if address.get(field) is not None:
addr_payload[field] = address[field]
@@ -443,7 +464,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
raise ValueError("emailAddress is required for email contacts")
payload["emailAddress"] = {
"emailAddress": email.get("emailAddress"),
- "emailPurpose": email.get("emailPurpose", "Email")
+ "emailPurpose": email.get("emailPurpose", "Email"),
}
else:
@@ -455,20 +476,15 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
headers = get_api_headers(context, optional_headers)
- response = await context.fetch(
- url,
- method="POST",
- json=payload,
- headers=headers
- )
+ response = await context.fetch(url, method="POST", json=payload, headers=headers)
return ActionResult(
data={
"contact": response,
"result": True,
- "message": "Contact added successfully"
+ "message": "Contact added successfully",
},
- cost_usd=None
+ cost_usd=None,
)
except Exception as e:
@@ -484,17 +500,17 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
"Call get_company_contacts first to get the addressId and etag, "
"then call update_company_contact with that contactId and etag."
),
- "error": error_str
+ "error": error_str,
},
- cost_usd=None
+ cost_usd=None,
)
return ActionResult(
data={
"result": False,
"message": f"Error adding company contact: {error_str}",
- "error": error_str
+ "error": error_str,
},
- cost_usd=None
+ cost_usd=None,
)
@@ -536,12 +552,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
headers = get_api_headers(context, optional_headers)
- response = await context.fetch(
- url,
- method="GET",
- params=params,
- headers=headers
- )
+ response = await context.fetch(url, method="GET", params=params, headers=headers)
addresses = response.get("items", []) if isinstance(response, dict) else response
@@ -550,9 +561,9 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
"addresses": addresses,
"count": len(addresses) if isinstance(addresses, list) else 0,
"searchType": "dpid" if dpid else "query",
- "result": True
+ "result": True,
},
- cost_usd=None
+ cost_usd=None,
)
except Exception as e:
@@ -560,9 +571,9 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
data={
"result": False,
"message": f"Error searching NZ addresses: {str(e)}",
- "error": str(e)
+ "error": str(e),
},
- cost_usd=None
+ cost_usd=None,
)
@@ -594,7 +605,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
"name": name,
"emailAddress": email_address,
"designation": designation,
- "companyDetailsConfirmedCorrectAsOfETag": etag
+ "companyDetailsConfirmedCorrectAsOfETag": etag,
}
# Optional: mobile phone for next year's SMS reminders
@@ -628,7 +639,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
if inputs.get("organisationId"):
payload["fileAnnualReturnForOrganisation"] = {
"organisationId": inputs["organisationId"],
- "name": inputs.get("organisationName", "")
+ "name": inputs.get("organisationName", ""),
}
optional_headers = {}
@@ -637,12 +648,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
headers = get_api_headers(context, optional_headers)
- response = await context.fetch(
- url,
- method="POST",
- json=payload,
- headers=headers
- )
+ response = await context.fetch(url, method="POST", json=payload, headers=headers)
is_credit_card = payment_method == "creditCard"
payment_info_resp = response.get("paymentInfo", {}) if isinstance(response, dict) else {}
@@ -661,9 +667,9 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
"Annual return filed. Complete payment at the paymentUrl to finalise."
if is_credit_card
else "Annual return filed successfully via direct debit."
- )
+ ),
},
- cost_usd=None
+ cost_usd=None,
)
except Exception as e:
@@ -671,7 +677,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
data={
"result": False,
"message": f"Error filing annual return: {str(e)}",
- "error": str(e)
+ "error": str(e),
},
- cost_usd=None
+ cost_usd=None,
)
diff --git a/companies-register/tests/context.py b/companies-register/tests/context.py
index 3f5838a6..54fc9251 100644
--- a/companies-register/tests/context.py
+++ b/companies-register/tests/context.py
@@ -2,5 +2,3 @@
import os
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
-
-from companies_register import companies_register
diff --git a/companies-register/tests/test_companies_register.py b/companies-register/tests/test_companies_register.py
index 1552a833..99058d28 100644
--- a/companies-register/tests/test_companies_register.py
+++ b/companies-register/tests/test_companies_register.py
@@ -7,10 +7,10 @@
# Configuration — replace with real credentials before running
# ---------------------------------------------------------------------------
# Subscription key from https://portal.api.business.govt.nz/
-SUBSCRIPTION_KEY = "your_subscription_key_here"
+SUBSCRIPTION_KEY = "your_subscription_key_here" # nosec B105
# OAuth access token from RealMe authentication (sandbox: use L_testuser)
-ACCESS_TOKEN = "your_oauth_token_here"
+ACCESS_TOKEN = "your_oauth_token_here" # nosec B105
# NZBN of a registered company to test against (status 50)
# Example NZBN from MBIE sandbox data
@@ -29,12 +29,11 @@
# Auth helper
# ---------------------------------------------------------------------------
+
def make_auth():
return {
"auth_type": "PlatformOauth2",
- "credentials": {
- "access_token": ACCESS_TOKEN
- }
+ "credentials": {"access_token": ACCESS_TOKEN},
}
@@ -42,15 +41,12 @@ def make_auth():
# Tests
# ---------------------------------------------------------------------------
+
async def test_get_company_details():
"""Get company by NZBN (registered company, status 50)."""
async with ExecutionContext(auth=make_auth()) as context:
try:
- result = await companies_register.execute_action(
- "get_company_details",
- {"companyUuid": TEST_NZBN},
- context
- )
+ result = await companies_register.execute_action("get_company_details", {"companyUuid": TEST_NZBN}, context)
print(f"\n[get_company_details] {result.data}")
assert result.data.get("result") is True
assert result.data.get("etag") is not None, "ETag missing — required for annual return"
@@ -69,12 +65,11 @@ async def test_get_company_details_by_uuid():
"""Get company by UUID (pre-incorporated company)."""
async with ExecutionContext(auth=make_auth()) as context:
try:
- result = await companies_register.execute_action(
- "get_company_details",
- {"companyUuid": TEST_UUID},
- context
+ result = await companies_register.execute_action("get_company_details", {"companyUuid": TEST_UUID}, context)
+ print(
+ f"\n[get_company_details by UUID] {result.data.get('companyName')}"
+ f" — status {result.data.get('companyStatusCode')}"
)
- print(f"\n[get_company_details by UUID] {result.data.get('companyName')} — status {result.data.get('companyStatusCode')}")
return result
except Exception as e:
print(f" ERROR: {e}")
@@ -86,11 +81,9 @@ async def test_get_company_contacts():
async with ExecutionContext(auth=make_auth()) as context:
try:
result = await companies_register.execute_action(
- "get_company_contacts",
- {"companyUuid": TEST_NZBN},
- context
+ "get_company_contacts", {"companyUuid": TEST_NZBN}, context
)
- print(f"\n[get_company_contacts]")
+ print("\n[get_company_contacts]")
assert result.data.get("result") is True
addresses = result.data.get("physicalOrPostalAddresses", [])
phones = result.data.get("phoneContacts", [])
@@ -100,7 +93,10 @@ async def test_get_company_contacts():
print(f" emails : {len(emails)}")
print(f" etag : {result.data.get('etag')}")
for addr in addresses:
- print(f" [{addr.get('addressPurpose')}] {addr.get('address1')}, {addr.get('address3')} — id: {addr.get('addressId')}")
+ print(
+ f" [{addr.get('addressPurpose')}] {addr.get('address1')},"
+ f" {addr.get('address3')} — id: {addr.get('addressId')}"
+ )
return result
except Exception as e:
print(f" ERROR: {e}")
@@ -114,14 +110,17 @@ async def test_search_nz_address():
result = await companies_register.execute_action(
"search_nz_address",
{"find": "Level 1 15 Stout Street Wellington", "limit": 5},
- context
+ context,
)
- print(f"\n[search_nz_address]")
+ print("\n[search_nz_address]")
assert result.data.get("result") is True
addresses = result.data.get("addresses", [])
print(f" found : {result.data.get('count')} addresses")
for addr in addresses[:3]:
- print(f" dpid={addr.get('dpid')} — {addr.get('address1')}, {addr.get('address3')} {addr.get('postCode')}")
+ print(
+ f" dpid={addr.get('dpid')} — {addr.get('address1')},"
+ f" {addr.get('address3')} {addr.get('postCode')}"
+ )
return result
except Exception as e:
print(f" ERROR: {e}")
@@ -132,12 +131,8 @@ async def test_search_nz_address_by_dpid():
"""Look up a specific NZ Post address by DPID."""
async with ExecutionContext(auth=make_auth()) as context:
try:
- result = await companies_register.execute_action(
- "search_nz_address",
- {"dpid": "1889019"},
- context
- )
- print(f"\n[search_nz_address by DPID]")
+ result = await companies_register.execute_action("search_nz_address", {"dpid": "1889019"}, context)
+ print("\n[search_nz_address by DPID]")
addresses = result.data.get("addresses", [])
if addresses:
a = addresses[0]
@@ -163,12 +158,12 @@ async def test_add_company_contact_address():
"address1": "Level 1, 15 Stout Street",
"address3": "Wellington",
"postCode": "6011",
- "countryCode": "NZ"
- }
+ "countryCode": "NZ",
+ },
},
- context
+ context,
)
- print(f"\n[add_company_contact — address]")
+ print("\n[add_company_contact — address]")
print(f" result : {result.data.get('result')}")
print(f" contact : {result.data.get('contact')}")
return result
@@ -189,12 +184,12 @@ async def test_add_company_contact_address_with_dpid():
"physicalOrPostalAddress": {
"addressType": "Physical",
"addressPurpose": "Registered Office Address",
- "dpid": "1889019"
- }
+ "dpid": "1889019",
+ },
},
- context
+ context,
)
- print(f"\n[add_company_contact — address with dpid]")
+ print("\n[add_company_contact — address with dpid]")
print(f" result : {result.data.get('result')}")
return result
except Exception as e:
@@ -213,12 +208,12 @@ async def test_add_company_contact_email():
"contactType": "email",
"emailAddress": {
"emailAddress": "admin@testcompany.co.nz",
- "emailPurpose": "Email"
- }
+ "emailPurpose": "Email",
+ },
},
- context
+ context,
)
- print(f"\n[add_company_contact — email]")
+ print("\n[add_company_contact — email]")
print(f" result : {result.data.get('result')}")
return result
except Exception as e:
@@ -249,12 +244,12 @@ async def test_update_company_contact():
"addressId": TEST_CONTACT_ID,
"addressType": "Physical",
"addressPurpose": "Registered Office Address",
- "dpid": "1889019"
- }
+ "dpid": "1889019",
+ },
},
- context
+ context,
)
- print(f"\n[update_company_contact]")
+ print("\n[update_company_contact]")
print(f" result : {result.data.get('result')}")
print(f" message : {result.data.get('message')}")
print(f" new etag : {result.data.get('etag')}")
@@ -280,11 +275,11 @@ async def test_file_annual_return_direct_debit():
"emailAddress": {"emailAddress": "jane.smith@testcompany.co.nz"},
"designation": "Director",
"companyDetailsConfirmedCorrectAsOfETag": TEST_ETAG,
- "paymentMethod": "directDebit"
+ "paymentMethod": "directDebit",
},
- context
+ context,
)
- print(f"\n[file_annual_return — directDebit]")
+ print("\n[file_annual_return — directDebit]")
print(f" result : {result.data.get('result')}")
print(f" message : {result.data.get('message')}")
print(f" documentId : {result.data.get('documentId')}")
@@ -317,12 +312,12 @@ async def test_file_annual_return_credit_card():
"phoneContact": {
"phoneNumber": "211234567",
"countryCode": "64",
- "phonePurpose": "Mobile"
- }
+ "phonePurpose": "Mobile",
+ },
},
- context
+ context,
)
- print(f"\n[file_annual_return — creditCard]")
+ print("\n[file_annual_return — creditCard]")
print(f" result : {result.data.get('result')}")
print(f" message : {result.data.get('message')}")
print(f" paymentUrl : {result.data.get('paymentUrl')}")
@@ -336,6 +331,7 @@ async def test_file_annual_return_credit_card():
# Full workflow test — get company → get contacts → update address
# ---------------------------------------------------------------------------
+
async def test_address_update_workflow():
"""
End-to-end workflow:
@@ -344,18 +340,16 @@ async def test_address_update_workflow():
3. Get company contacts (get addressId + contacts etag)
4. Update the registered office address
"""
- print(f"\n{'='*60}")
+ print(f"\n{'=' * 60}")
print("WORKFLOW: Update Registered Office Address")
- print('='*60)
+ print("=" * 60)
async with ExecutionContext(auth=make_auth()) as context:
try:
# Step 1: Get company details
print("\nStep 1: Get company details...")
details = await companies_register.execute_action(
- "get_company_details",
- {"companyUuid": TEST_NZBN},
- context
+ "get_company_details", {"companyUuid": TEST_NZBN}, context
)
company_name = details.data.get("companyName")
print(f" Company: {company_name}")
@@ -365,7 +359,7 @@ async def test_address_update_workflow():
addr_search = await companies_register.execute_action(
"search_nz_address",
{"find": "15 Stout Street Wellington", "limit": 3},
- context
+ context,
)
addresses = addr_search.data.get("addresses", [])
if not addresses:
@@ -378,15 +372,13 @@ async def test_address_update_workflow():
# Step 3: Get contacts to find addressId
print("\nStep 3: Get company contacts...")
contacts_result = await companies_register.execute_action(
- "get_company_contacts",
- {"companyUuid": TEST_NZBN},
- context
+ "get_company_contacts", {"companyUuid": TEST_NZBN}, context
)
contacts_etag = contacts_result.data.get("etag")
physical_addresses = contacts_result.data.get("physicalOrPostalAddresses", [])
registered_office = next(
(a for a in physical_addresses if a.get("addressPurpose") == "Registered Office Address"),
- None
+ None,
)
if not registered_office:
print(" No Registered Office Address found — skipping update")
@@ -412,10 +404,10 @@ async def test_address_update_workflow():
"address3": addr.get("address3"),
"postCode": addr.get("postCode"),
"countryCode": "NZ",
- "effectiveDate": "2026-03-10T00:00:00Z"
- }
+ "effectiveDate": "2026-03-10T00:00:00Z",
+ },
},
- context
+ context,
)
print(f" result : {update_result.data.get('result')}")
print(f" message : {update_result.data.get('message')}")
@@ -430,6 +422,7 @@ async def test_address_update_workflow():
# Test runner
# ---------------------------------------------------------------------------
+
async def run_all_tests():
print("=" * 60)
print("Companies Register Integration Tests")
@@ -440,18 +433,18 @@ async def run_all_tests():
failed = 0
tests = [
- ("Get Company Details (NZBN)", test_get_company_details),
- ("Get Company Details (UUID)", test_get_company_details_by_uuid),
- ("Get Company Contacts", test_get_company_contacts),
- ("Search NZ Address (query)", test_search_nz_address),
- ("Search NZ Address (DPID)", test_search_nz_address_by_dpid),
- ("Add Contact — address (lines)", test_add_company_contact_address),
- ("Add Contact — address (dpid)", test_add_company_contact_address_with_dpid),
- ("Add Contact — email", test_add_company_contact_email),
- ("Update Contact", test_update_company_contact),
- ("File Annual Return (directDebit)", test_file_annual_return_direct_debit),
- ("File Annual Return (creditCard)", test_file_annual_return_credit_card),
- ("Workflow: Address Update", test_address_update_workflow),
+ ("Get Company Details (NZBN)", test_get_company_details),
+ ("Get Company Details (UUID)", test_get_company_details_by_uuid),
+ ("Get Company Contacts", test_get_company_contacts),
+ ("Search NZ Address (query)", test_search_nz_address),
+ ("Search NZ Address (DPID)", test_search_nz_address_by_dpid),
+ ("Add Contact — address (lines)", test_add_company_contact_address),
+ ("Add Contact — address (dpid)", test_add_company_contact_address_with_dpid),
+ ("Add Contact — email", test_add_company_contact_email),
+ ("Update Contact", test_update_company_contact),
+ ("File Annual Return (directDebit)", test_file_annual_return_direct_debit),
+ ("File Annual Return (creditCard)", test_file_annual_return_credit_card),
+ ("Workflow: Address Update", test_address_update_workflow),
]
for name, fn in tests:
@@ -467,7 +460,7 @@ async def run_all_tests():
print(f" FAIL {name} — {e}")
failed += 1
- print(f"\n{'='*60}")
+ print(f"\n{'=' * 60}")
print(f"Results: {passed} passed, {failed} failed")
print("=" * 60)
diff --git a/fathom/fathom.py b/fathom/fathom.py
index f93425e3..9079f440 100644
--- a/fathom/fathom.py
+++ b/fathom/fathom.py
@@ -1,11 +1,15 @@
from autohive_integrations_sdk import (
- Integration, ExecutionContext, ActionHandler, ActionResult
+ Integration,
+ ExecutionContext,
+ ActionHandler,
+ ActionResult,
)
from typing import Dict, Any, Optional
from urllib.parse import quote
fathom = Integration.load()
+
class FathomAPIClient:
"""Client for interacting with the Fathom API"""
@@ -13,13 +17,17 @@ def __init__(self, context: ExecutionContext):
self.context = context
self.base_url = "https://api.fathom.ai/external/v1"
- async def _make_request(self, endpoint: str, method: str = "GET", params: Optional[Dict] = None, data: Optional[Dict] = None):
+ async def _make_request(
+ self,
+ endpoint: str,
+ method: str = "GET",
+ params: Optional[Dict] = None,
+ data: Optional[Dict] = None,
+ ):
"""Make an authenticated request to the Fathom API"""
url = f"{self.base_url}/{endpoint}"
- headers = {
- "Content-Type": "application/json"
- }
+ headers = {"Content-Type": "application/json"}
# Build query string manually for array parameters
if params and method == "GET":
@@ -50,8 +58,10 @@ async def _make_request(self, endpoint: str, method: str = "GET", params: Option
else:
raise ValueError(f"Unsupported HTTP method: {method}")
+
# ---- Action Handlers ----
+
@fathom.action("list_meetings")
class ListMeetingsAction(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
@@ -92,7 +102,12 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
processed_items = []
# Fields to exclude when they are null (not in output schema)
- optional_fields = ["transcript", "action_items", "default_summary", "crm_matches"]
+ optional_fields = [
+ "transcript",
+ "action_items",
+ "default_summary",
+ "crm_matches",
+ ]
for item in items:
processed_item = {}
@@ -108,21 +123,17 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
data={
"limit": response.get("limit", 0),
"next_cursor": response.get("next_cursor"),
- "items": processed_items
+ "items": processed_items,
},
- cost_usd=0.0
+ cost_usd=0.0,
)
except Exception as e:
return ActionResult(
- data={
- "limit": 0,
- "next_cursor": None,
- "items": [],
- "error": str(e)
- },
- cost_usd=0.0
+ data={"limit": 0, "next_cursor": None, "items": [], "error": str(e)},
+ cost_usd=0.0,
)
+
@fathom.action("get_transcript")
class GetTranscriptAction(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
@@ -138,29 +149,25 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
speaker = segment.get("speaker", {})
speaker_name = speaker.get("display_name", "Unknown Speaker")
- transcript.append({
- "speaker_name": speaker_name,
- "timestamp": segment.get("timestamp", "00:00:00"),
- "text": segment.get("text", "")
- })
+ transcript.append(
+ {
+ "speaker_name": speaker_name,
+ "timestamp": segment.get("timestamp", "00:00:00"),
+ "text": segment.get("text", ""),
+ }
+ )
return ActionResult(
- data={
- "recording_id": recording_id,
- "transcript": transcript
- },
- cost_usd=0.0
+ data={"recording_id": recording_id, "transcript": transcript},
+ cost_usd=0.0,
)
except Exception as e:
return ActionResult(
- data={
- "recording_id": recording_id,
- "transcript": [],
- "error": str(e)
- },
- cost_usd=0.0
+ data={"recording_id": recording_id, "transcript": [], "error": str(e)},
+ cost_usd=0.0,
)
+
@fathom.action("list_teams")
class ListTeamsAction(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
@@ -176,30 +183,28 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
teams = []
for team in response.get("items", []):
- teams.append({
- "name": team.get("name", ""),
- "created_at": team.get("created_at", "")
- })
+ teams.append(
+ {
+ "name": team.get("name", ""),
+ "created_at": team.get("created_at", ""),
+ }
+ )
return ActionResult(
data={
"limit": response.get("limit"),
"next_cursor": response.get("next_cursor"),
- "teams": teams
+ "teams": teams,
},
- cost_usd=0.0
+ cost_usd=0.0,
)
except Exception as e:
return ActionResult(
- data={
- "limit": None,
- "next_cursor": None,
- "teams": [],
- "error": str(e)
- },
- cost_usd=0.0
+ data={"limit": None, "next_cursor": None, "teams": [], "error": str(e)},
+ cost_usd=0.0,
)
+
@fathom.action("list_team_members")
class ListTeamMembersAction(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
@@ -217,19 +222,21 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
team_members = []
for member in response.get("items", []):
- team_members.append({
- "name": member.get("name", ""),
- "email": member.get("email", ""),
- "created_at": member.get("created_at", "")
- })
+ team_members.append(
+ {
+ "name": member.get("name", ""),
+ "email": member.get("email", ""),
+ "created_at": member.get("created_at", ""),
+ }
+ )
return ActionResult(
data={
"limit": response.get("limit"),
"next_cursor": response.get("next_cursor"),
- "team_members": team_members
+ "team_members": team_members,
},
- cost_usd=0.0
+ cost_usd=0.0,
)
except Exception as e:
return ActionResult(
@@ -237,7 +244,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
"limit": None,
"next_cursor": None,
"team_members": [],
- "error": str(e)
+ "error": str(e),
},
- cost_usd=0.0
+ cost_usd=0.0,
)
diff --git a/fathom/tests/context.py b/fathom/tests/context.py
index 70659667..4e97343e 100644
--- a/fathom/tests/context.py
+++ b/fathom/tests/context.py
@@ -1,7 +1,6 @@
# -*- coding: utf-8 -*-
import sys
import os
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../dependencies")))
-from fathom import fathom
\ No newline at end of file
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../dependencies")))
diff --git a/fathom/tests/test_fathom.py b/fathom/tests/test_fathom.py
index 46dbcf3b..1147b1c1 100644
--- a/fathom/tests/test_fathom.py
+++ b/fathom/tests/test_fathom.py
@@ -3,14 +3,13 @@
from context import fathom
from autohive_integrations_sdk import ExecutionContext
+
async def test_list_meetings():
"""Test listing meetings"""
print("\n--- Testing list_meetings ---")
# Setup mock auth object
- auth = {
- "api_key": "test_api_key"
- }
+ auth = {"api_key": "test_api_key"}
inputs = {}
@@ -24,17 +23,14 @@ async def test_list_meetings():
except Exception as e:
print(f"✗ Error testing list_meetings: {str(e)}")
+
async def test_get_transcript():
"""Test getting recording transcript"""
print("\n--- Testing get_transcript ---")
- auth = {
- "api_key": "test_api_key"
- }
+ auth = {"api_key": "test_api_key"}
- inputs = {
- "recording_id": "test_recording_id_123"
- }
+ inputs = {"recording_id": "test_recording_id_123"}
async with ExecutionContext(auth=auth) as context:
try:
@@ -43,13 +39,12 @@ async def test_get_transcript():
except Exception as e:
print(f"✗ Error testing get_transcript: {str(e)}")
+
async def test_list_teams():
"""Test listing teams"""
print("\n--- Testing list_teams ---")
- auth = {
- "api_key": "test_api_key"
- }
+ auth = {"api_key": "test_api_key"}
inputs = {}
@@ -60,13 +55,12 @@ async def test_list_teams():
except Exception as e:
print(f"✗ Error testing list_teams: {str(e)}")
+
async def test_list_team_members():
"""Test listing team members"""
print("\n--- Testing list_team_members ---")
- auth = {
- "api_key": "test_api_key"
- }
+ auth = {"api_key": "test_api_key"}
inputs = {}
@@ -77,6 +71,7 @@ async def test_list_team_members():
except Exception as e:
print(f"✗ Error testing list_team_members: {str(e)}")
+
async def main():
print("========================================")
print("Testing Fathom Integration")
@@ -91,5 +86,6 @@ async def main():
print("Tests completed!")
print("========================================")
+
if __name__ == "__main__":
asyncio.run(main())
diff --git a/ghost/ghost.py b/ghost/ghost.py
index 3151f3db..58fbb18b 100644
--- a/ghost/ghost.py
+++ b/ghost/ghost.py
@@ -25,9 +25,7 @@ def _get_base_url(context: ExecutionContext) -> str:
return url.rstrip("/")
-def _content_get(
- context: ExecutionContext, endpoint: str, params: Optional[Dict] = None
-) -> Dict:
+def _content_get(context: ExecutionContext, endpoint: str, params: Optional[Dict] = None) -> Dict:
credentials = context.auth.get("credentials", {})
content_key = credentials.get("content_api_key", "")
if not content_key:
@@ -71,14 +69,10 @@ def _admin_request(
token = _make_admin_jwt(context)
headers = {"Authorization": f"Ghost {token}"}
if files:
- response = requests.request(
- method, url, headers=headers, files=files, params=params, timeout=30
- )
+ response = requests.request(method, url, headers=headers, files=files, params=params, timeout=30)
else:
headers["Content-Type"] = "application/json"
- response = requests.request(
- method, url, headers=headers, json=json, params=params, timeout=30
- )
+ response = requests.request(method, url, headers=headers, json=json, params=params, timeout=30)
response.raise_for_status()
return response.json() if response.content else {}
@@ -108,9 +102,7 @@ def _parse_error(e: Exception) -> tuple:
def _error(e: Exception) -> ActionResult:
error_msg, error_type = _parse_error(e)
- return ActionResult(
- data={"result": False, "error": error_msg, "error_type": error_type}
- )
+ return ActionResult(data={"result": False, "error": error_msg, "error_type": error_type})
# ---- Content API Actions ----
@@ -122,16 +114,10 @@ class GetPostsAction(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
- params = {
- k: inputs[k]
- for k in ("limit", "page", "filter", "include")
- if inputs.get(k)
- }
+ params = {k: inputs[k] for k in ("limit", "page", "filter", "include") if inputs.get(k)}
params.setdefault("limit", 15)
data = _content_get(context, "posts", params)
- return _success(
- {"posts": data.get("posts", []), "meta": data.get("meta", {})}
- )
+ return _success({"posts": data.get("posts", []), "meta": data.get("meta", {})})
except Exception as e:
return _error(e)
@@ -165,14 +151,10 @@ class GetPagesAction(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
- params = {
- k: inputs[k] for k in ("limit", "page", "filter") if inputs.get(k)
- }
+ params = {k: inputs[k] for k in ("limit", "page", "filter") if inputs.get(k)}
params.setdefault("limit", 15)
data = _content_get(context, "pages", params)
- return _success(
- {"pages": data.get("pages", []), "meta": data.get("meta", {})}
- )
+ return _success({"pages": data.get("pages", []), "meta": data.get("meta", {})})
except Exception as e:
return _error(e)
@@ -206,14 +188,10 @@ class GetTagsAction(ActionHandler):
async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
try:
- params = {
- k: inputs[k] for k in ("limit", "page", "filter") if inputs.get(k)
- }
+ params = {k: inputs[k] for k in ("limit", "page", "filter") if inputs.get(k)}
params.setdefault("limit", 15)
data = _content_get(context, "tags", params)
- return _success(
- {"tags": data.get("tags", []), "meta": data.get("meta", {})}
- )
+ return _success({"tags": data.get("tags", []), "meta": data.get("meta", {})})
except Exception as e:
return _error(e)
@@ -227,9 +205,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
params = {k: inputs[k] for k in ("limit", "page") if inputs.get(k)}
params.setdefault("limit", 15)
data = _content_get(context, "authors", params)
- return _success(
- {"authors": data.get("authors", []), "meta": data.get("meta", {})}
- )
+ return _success({"authors": data.get("authors", []), "meta": data.get("meta", {})})
except Exception as e:
return _error(e)
@@ -281,9 +257,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
post[field] = inputs[field]
post.setdefault("status", "draft")
params = {"source": "html"} if inputs.get("html") else None
- data = _admin_request(
- context, "POST", "posts", json={"posts": [post]}, params=params
- )
+ data = _admin_request(context, "POST", "posts", json={"posts": [post]}, params=params)
posts = data.get("posts", [])
return _success({"post": posts[0] if posts else None})
except Exception as e:
@@ -336,9 +310,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
page[field] = inputs[field]
page.setdefault("status", "draft")
params = {"source": "html"} if inputs.get("html") else None
- data = _admin_request(
- context, "POST", "pages", json={"pages": [page]}, params=params
- )
+ data = _admin_request(context, "POST", "pages", json={"pages": [page]}, params=params)
pages = data.get("pages", [])
return _success({"page": pages[0] if pages else None})
except Exception as e:
@@ -378,9 +350,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
for field in ["name", "labels", "newsletters", "note"]:
if inputs.get(field) is not None:
member[field] = inputs[field]
- data = _admin_request(
- context, "POST", "members", json={"members": [member]}
- )
+ data = _admin_request(context, "POST", "members", json={"members": [member]})
members = data.get("members", [])
return _success({"member": members[0] if members else None})
except Exception as e:
@@ -398,9 +368,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
for field in ["email", "name", "labels", "newsletters", "note"]:
if inputs.get(field) is not None:
member[field] = inputs[field]
- data = _admin_request(
- context, "PUT", f"members/{member_id}", json={"members": [member]}
- )
+ data = _admin_request(context, "PUT", f"members/{member_id}", json={"members": [member]})
members = data.get("members", [])
return _success({"member": members[0] if members else None})
except Exception as e:
diff --git a/ghost/tests/test_ghost.py b/ghost/tests/test_ghost.py
index 7c222b0f..56950bc3 100644
--- a/ghost/tests/test_ghost.py
+++ b/ghost/tests/test_ghost.py
@@ -174,9 +174,7 @@ async def test_update_post(post):
)
data = get_data(result)
assert data.get("result") is True, f"Expected result=True, got: {data}"
- assert data["post"]["title"] == "Test Post from Autohive (updated)", (
- "Title not updated"
- )
+ assert data["post"]["title"] == "Test Post from Autohive (updated)", "Title not updated"
print(f"OK — updated post {data['post']['id']}")
return data["post"]
diff --git a/jira/jira.py b/jira/jira.py
index 26ee4047..4b561e6f 100644
--- a/jira/jira.py
+++ b/jira/jira.py
@@ -365,7 +365,14 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext):
params = {} if notify_users else {"notifyUsers": "false"}
url = api_url(cloud_id, f"/issue/{safe_path(issue_key)}")
- await jira_request("PUT", url, access_token, context, params=params or None, payload=payload)
+ await jira_request(
+ "PUT",
+ url,
+ access_token,
+ context,
+ params=params or None,
+ payload=payload,
+ )
return ActionResult(
data={