Catalog API Invoke API
Summary
The invoke catalog API is used to upload items to the Amazon Just WalkOut API.
Note: Make sure to follow your organization's process to prepare the code for production release.
Python Code
import requests
from requests_aws4auth import AWS4Auth
import json
import boto3
import os
def update_catalog():
try:
# Load catalog JSON from a file or environment variable
with open('catalog.json', 'r') as f:
request_parameters = json.load(f)
# Get role ARN from environment variable
role_arn = os.environ['CATALOG_API_ROLE_ARN']
# Get temp credentials using STS
client_sts = boto3.client('sts')
response = client_sts.assume_role(
RoleArn=role_arn,
RoleSessionName='CatalogAPI'
)
credentials = response['Credentials']
# Auth the outgoing API call
auth = AWS4Auth(
credentials['AccessKeyId'],
credentials['SecretAccessKey'],
os.environ['AWS_REGION'],
'execute-api',
session_token=credentials['SessionToken']
)
# Use requests library to post to the API end point
api_url = os.environ['CATALOG_API_ENDPOINT']
response = requests.post(api_url, auth=auth, json=request_parameters)
# Check response code and act accordingly
if response.status_code == 201:
response_text = response.json()
print(f"Ingestion ID: {response_text['ingestionId']}")
elif response.status_code == 403:
print("Error authenticating to the API")
elif response.status_code == 400:
print("Error: Bad request")
else:
print(f"Unexpected status code: {response.status_code}")
print(f"Response status code: {response.status_code}")
print(f"Response body: {response.text}")
except requests.exceptions.RequestException as e:
print(f"Request failed: {e}")
except json.JSONDecodeError:
print("Failed to parse response JSON")
except Exception as e:
print(f"An unexpected error occurred: {e}")
Unit Tests
import pytest
import json
import requests
from unittest.mock import patch, MagicMock
from botocore.exceptions import ClientError
from src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function import (
lambda_handler,
sign_request,
validate_payload,
get_table
)
@pytest.fixture
def valid_event():
return {
"catalogItems": [
{
"item_sku": "Test-012000161155",
"external_product_id": "012000161155",
"external_product_id_type": "UPC",
"item_name": "Life Water",
"store_id": "SAMPLE_STORE",
"standard_price": "1.00",
"brand_name": "Life Water",
"product_tax_code": "A_GEN_TAX",
"product_category": "Drink",
"product_subcategory": "Water"
}
]
}
@pytest.fixture
def mock_table():
return MagicMock()
@pytest.fixture
def mock_dynamodb(mock_table):
with patch('boto3.resource') as mock_dynamo:
mock_dynamo.return_value.Table.return_value = mock_table
yield mock_dynamo
class TestLambdaFunction:
@patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.requests.post')
@patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.sign_request')
@patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.get_table')
def test_successful_execution(self, mock_get_table, mock_sign, mock_post, valid_event):
mock_table = MagicMock()
mock_get_table.return_value = mock_table
mock_response = MagicMock()
mock_response.json.return_value = {"status": "success"}
mock_response.raise_for_status.return_value = None
mock_post.return_value = mock_response
mock_sign.return_value = {"Authorization": "test-auth"}
response = lambda_handler(valid_event, None)
assert response['statusCode'] == 200
assert 'Successfully processed catalog items' in response['body']
mock_table.put_item.assert_called_once()
mock_post.assert_called_once()
@patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.requests.post')
@patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.sign_request')
def test_api_error(self, mock_sign, mock_post, valid_event):
mock_error = requests.exceptions.RequestException("API Error")
mock_post.side_effect = mock_error
mock_sign.return_value = {"Authorization": "test-auth"}
response = lambda_handler(valid_event, None)
assert response['statusCode'] == 500
assert 'Error processing catalog items: API Error' in response['body']
def test_missing_api_url(self, monkeypatch, valid_event):
monkeypatch.delenv('API_URL', raising=False)
response = lambda_handler(valid_event, None)
assert response['statusCode'] == 500
assert 'API_URL environment variable is not set' in response['body']
@patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.get_table')
def test_unexpected_error(self, mock_get_table, valid_event):
mock_get_table.side_effect = ClientError(
error_response={'Error': {'Code': 'InternalServerError'}},
operation_name='GetTable'
)
response = lambda_handler(valid_event, None)
assert response['statusCode'] == 500
assert 'Unexpected error processing catalog items' in response['body']
mock_get_table.assert_called_once()
def test_validate_payload_valid(self, valid_event):
assert validate_payload(valid_event) is True
@pytest.mark.parametrize("invalid_payload", [
None,
{},
{"catalogItems": []},
{"catalogItems": [{}]},
{"catalogItems": [{"item_sku": "test"}]},
])
def test_validate_payload_invalid(self, invalid_payload):
assert validate_payload(invalid_payload) is False
@patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.boto3.Session')
def test_sign_request(self, mock_session):
mock_credentials = MagicMock(
access_key='test-key',
secret_key='test-secret',
token='test-token'
)
mock_session.return_value.get_credentials.return_value = mock_credentials
url = 'https://api-test.amazonaws.com/dev/catalog'
method = 'POST'
body = {"test": "data"}
headers = sign_request(url, method, body)
assert isinstance(headers, dict)
assert 'Content-Type' in headers
assert headers['Content-Type'] == 'application/json'
mock_session.return_value.get_credentials.assert_called_once()
def test_get_table(self, mock_dynamodb):
table = get_table()
assert table is not None
mock_dynamodb.assert_called_once_with('dynamodb')
mock_dynamodb.return_value.Table.assert_called_once_with('CatalogAPICalls')
@patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.requests.post')
@patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.sign_request')
@patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.get_table')
def test_lambda_handler_invalid_payload(self, mock_get_table, mock_sign, mock_post):
response = lambda_handler({"invalid": "payload"}, None)
assert response['statusCode'] == 400
assert 'Invalid payload structure' in response['body']
@patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.requests.post')
@patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.sign_request')
@patch('src.asset_catalog_api_solution.lambda_functions.invoke_api_lambda_function.get_table')
def test_lambda_handler_empty_event(self, mock_get_table, mock_sign, mock_post):
response = lambda_handler({}, None)
assert response['statusCode'] == 400
assert 'Invalid payload structure' in response['body']