
Background
AWS API Gateway is compromised of an API instance creator and a running instance. The running instance is created when the user hit 'Create API' button in the AWS Console.
Once the instance is created, AWS API Gateway provides a URL that we can hit. If we hit it immediately it will return a 404.
So the next step is to configure the running instance to handle requests. AWS provides multiple ways to configure it, through cli, apis, CloudFormation templates or AWS Console UI. Checkout this link for more information on how to configure AWS API Gateway.
Configuring a Stage, a Route and an Integration is the minimum steps required to make the instance do something "useful".
In this article, I'll focus on the running instance For this to work we need to configure a stage, a route and an integration. Configurations are just plain old JSON files, which we'll go through in the next section.
Let's get started!
Configurations
Starting with an example: We want to map http://localhost:8000/v0/postman to https://postman-echo.com/get and get the same results.
That's it! For this to work on AWS, we need to configure the following:
- define an HTTP_PROXY 'Integration' pointing to https://postman-echo.com/get.
- define a 'Route'; maping GET request to the integration created in previous step.
- define a 'Stage' named 'v0' with 'AutoDeploy' enabled.
We can do this using the AWS Console, a CloudFormation template or using CDK. Pick your tool of choice and try it out.
We'll do something similar for out API instance. Create JSON files to represent stages, routes and integrations.
So, I'll create three directories named stages
, routes
,
integrations
. Eeach folder can hold multiple
configurations. However, they'll hold a single file for the purpose of
this article.
Integrations
In AWS API Gateway, there are multiple integration types that can be defined. Such as HTTP_PROXY, integrating with AWS Lambda Function, etc. For this example we'll use HTTP_PROXY. Which is just a pass through proxy.
Let's create a file called pa09itl.json
inside the integrations/
folder, pa09itl
is considered the logical id of the resource.
And we'll define the integration object:
{
"IntegrationType": "HTTP_PROXY",
"IntegrationUri": "https://postman-echo.com/get"
}
Routes
A route is a path mapper to a target. Routes don't have names, so we'll generate a random id. AWS refers to these IDs as logical ID.
So, inside the routes/
directory we'll create a file named
ja32tiv.json
with the following content:
{
"RouteKey": "GET /postman",
"Target": "integrations/pa09itl"
}
Stages
A stage is just a named collection of routes. <domain>/{stage_name}/<path>.
Creating a file inside the stages/
folder named the same as the stage
name, v0.json
with the following content:
{
"Name": "v0",
"Routes": ["routes/ja32tiv"]
}
Implementation
Now we have configured the instance with stages, routes and integrations. All we need to do is to start handling requests. We need an HTTP server listening on a port; handling requests.
I'll be using Docker, Python and Uvicorn to implement this.
async def app(scope, receive, send):
try:
assert scope['type'] == 'http'
stage, path = load_stage_and_routes_for(path=scope['path'])
route = find_route(stage, method=scope['method'], path=path)
await invoke(route['Target'], scope, receive, send)
except Exception as e:
import traceback
traceback.print_exc()
await send({
'type': 'http.response.start',
'status': 404,
'headers': [
[b'content-type', b'application/json'],
],
})
await send({
'type': 'http.response.body',
'body': b'{"message":"Not Found"}'
})
def load_stage_and_routes_for(path):
stage_name = path.split('/')[1]
stage_path = f'stages/{stage_name}'
default_path = 'stages/default'
def load_from_file(json_path):
json_path = f'{json_path}.json'
if os.path.exists(json_path):
with open(json_path, 'r', encoding='utf8') as fd:
return json.load(fd)
selected_stage = load_from_file(stage_path) or load_from_file(default_path)
if selected_stage is None:
raise Exception('No stage defined')
selected_stage['Routes'] = map(load_from_file, selected_stage['Routes'])
path_without_stage = path.replace(f'/{stage_name}', '')
return (selected_stage, path_without_stage)
def find_route(stage, method, path):
routes = {_['RouteKey']: _ for _ in stage['Routes']}
match = route_selector.select_route(routes.keys(), method, path)
return routes[match]
async def invoke(target, scope, recieve, send):
with open(f'{target}.json', 'r', encoding='utf8') as fd:
t = json.load(fd)
if t['IntegrationType'] != 'HTTP_PROXY':
raise Exception('Not Supported')
with urllib.request.urlopen(t['IntegrationUri']) as response:
data = response.read()
await send({
'type': 'http.response.start',
'status': 200,
'headers': [
[b'content-type', b'application/json'],
],
})
await send({
'type': 'http.response.body',
'body': data
})
from . import route_selector
import json
import os
import urllib.request
route_selector
function is a different topic that I've covered here
https://www.nanosn.com/posts/aws-api-gateway-routing-incoming-requests-select-route/readme/
End to End Testing
import pytest
from multiprocessing import Process
import uvicorn
import requests
import time
import apigateway.main
@pytest.fixture
def start_server():
proc = Process(
target=uvicorn.run,
args=(apigateway.main.app,),
kwargs={
"host": "127.0.0.1",
"port": 5000,
"log_level": "info"},
daemon=True
)
proc.start()
time.sleep(0.1)
yield proc
proc.terminate()
def test_apigateway(start_server):
actual = requests.get('http://localhost:5000/v0/postman').json()
expected = requests.get('https://postman-echo.com/get').json()
del actual['headers']
del expected['headers']
assert actual == expected
And to run the tests:
docker run --name test_apigateway -td --rm -v `pwd`:/work --workdir /work python sleep infinity
docker exec -it test_apigateway pip install -e .[test]
docker exec -it test_apigateway pytest -vs tests/main_test.py
docker rm -f test_apigateway
Appendex A: Missing Files to Make the Application Run
[project]
name = "apigateway"
dependencies = [
'uvicorn',
'requests'
]
version = "0.0.1"
authors = [
{ name="NanoSN LLC.", email="info@nanosn.com" },
]
description = "Sample Implementation of AWS API Gateway"
requires-python = ">=3.10"
classifiers = [
"Programming Language :: Python :: 3",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
]
[project.optional-dependencies]
test = [
'pytest',
'pytest-cov'
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
# Required for relative imports.
def select_route(route_keys, request_method, request_path):
return (
find_with_exact_match(route_keys, request_method, request_path) or
find_with_greedy_match(route_keys, request_method, request_path) or
'$default'
)
def find_with_exact_match(route_keys, request_method, request_path):
route = f'{request_method} {request_path}'
if route in route_keys:
return route
request_path_parts = request_path.split('/')
len_request_path_parts = len(request_path_parts)
possible_matches = []
for route_key in route_keys:
method, path = route_key.split(' ')
key_path_parts = path.split('/')
if len_request_path_parts != len(key_path_parts):
continue
if ((is_possible_match(request_path_parts, key_path_parts)) and
(method == request_method or method == 'ANY')):
return route_key
def find_with_greedy_match(route_keys, request_method, request_path):
greedy_route_keys = [_ for _ in route_keys if is_greedy_variable(_)]
request_path_parts = request_path.split('/')
len_request_path_parts = len(request_path_parts)
for route_key in greedy_route_keys:
method, path = route_key.split(' ')
key_path_parts = path.split('/')
if ((is_possible_greedy_match(request_path_parts, key_path_parts)) and
(method == request_method or method == 'ANY')):
return route_key
def is_possible_match(request_path_parts, key_path_parts):
for index, _ in enumerate(key_path_parts):
if is_variable(key_path_parts[index]): # '{var}'
continue
if key_path_parts[index] != request_path_parts[index]:
return False
return True
def is_possible_greedy_match(request_path_parts, key_path_parts):
for index, _ in enumerate(key_path_parts):
if is_greedy_variable(key_path_parts[index]):
return True
if key_path_parts[index] != request_path_parts[index]:
return False
return False
def is_variable(part):
return part.startswith('{') and part.endswith('}')
def is_greedy_variable(part):
return part.startswith('{') and part.endswith('+}')