Implementing AWS API Gateway from Scratch: Create API Instance

../intro.png — Internals of AWS API Gateway (Partial)

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.

../create_api.png

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"
}

integrations/pa09itl.json

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"
}

routes/ja32tiv.json

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"]
}

stages/v0.json

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

src/apigateway/main.py

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

tests/main_test.py

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

run_tests.sh

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"

pyproject.toml

# Required for relative imports.

src/apigateway/__init__.py


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('+}')

src/apigateway/route_selector.py