Using a Python HTTP Proxy Server to Simulate an API Gateway

Our software development team uses Azure API Management as an API Gateway. This API Gateway is responsible for forwarding requests to the appropriate backend service. It also applies various inbound and outbound policies, such as validating the presence of a well formed JWT, IP filtering, and attaching request/response headers.

When integration testing our API backends, we want to use a single base URL when making requests to the API. (This is as opposed to keeping track of which endpoints point to which backend in the API client itself.) We also want to ensure that any HTTP headers the API backend expects to have been provided through the API Gateway have been populated.

To achieve this, we wrote a pretty simple HTTP Proxy server in Python. Here, I’ll walk through creating and running a simplified version of our Python proxy server.

Creating an Initial Proxy Server

We will extend the BaseHTTPRequestHandler from http.server to create our HTTP Server as follows:

import requests
import signal
import sys
import urllib.parse
from http.server import BaseHTTPRequestHandler, HTTPServer

class ApiProxy:
    def start_server(self):
        class ProxyHTTPRequestHandler(BaseHTTPRequestHandler):
            protocol_version = "HTTP/1.0"

            def do_GET(self):
                self._handle_request("get", requests.get)

            def do_DELETE(self):
                self._handle_request("delete", requests.delete)

            def do_POST(self):
                self._handle_request("post", requests.post)

            def do_PUT(self):
                self._handle_request("put", requests.put)

            def do_PATCH(self):
                self._handle_request("patch", requests.patch)

            def _handle_request(self, method, requests_func):
                url = self._resolve_url()
                if (url is None):
                    print(f"Unable to resolve the URL {self.path}")
                    self.send_response(404)
                    self.send_header("Content-Type", "application/json")
                    self.end_headers()
                    return

                body = self.rfile.read(int(self.headers["content-length"]))
                headers = dict(self.headers)

                # Set any custom headers that would be provided through API Management inbound policies
                headers['Api-Version'] = 'v1'
                headers['Api-Path'] = '/internal'
		
                resp = requests_func(url, data=body, headers=headers)

                self.send_response(resp.status_code)
                for key in headers:
                    self.send_header(key, headers[key])
                self.end_headers()
                self.wfile.write(resp.content)

            def _resolve_url(self):
                return "resolved url goes here"

        server_address = ('', 8001)
        self.httpd = HTTPServer(server_address, ProxyHTTPRequestHandler)
        print('proxy server is running')
        self.httpd.serve_forever()


def exit_now(signum, frame):
    sys.exit(0)

if __name__ == '__main__':
    proxy = ApiProxy()
    signal.signal(signal.SIGTERM, exit_now)
    proxy.start_server()

Each HTTP method will point to a common request handler that sends the request to the appropriate backend URL and passes back the response.

The _handle_request function will also set additional custom request headers. In this example, we update the headers with API versions and path prefixes. You can use these, say, when constructing locations for created responses.

Mapping Routes

Next, we will implement the _resolve_url function to map requests to their appropriate backend API handlers. We use routes.mapper to group endpoints sharing a common backend together. In this simplified example, we have two services, “users” and “organizations”, so we add the following to the top of the file:

from routes import Mapper

map = Mapper()

def route(mapper, path_template):
    m.connect(None, path_template, action=path_template)

with map.submapper(controller="users") as m:
    route(m, "/users")
    route(m, "/users/{userId}")

with map.submapper(controller="organizations") as m:
    route(m, "/organizations")
    route(m, "/organizations/{organizationId}")

With our mapper set up, we can now implement _resolve_url.

            def _resolve_url(self):
                parts = urllib.parse.urlparse(self.path)

                path = parts.path
                match = map.match(path)

                if (match is None):
                    return None
		    
                service = match['controller']
                base_url = get_base_url_for_service(service)

                new_url = f"{base_url}{self.path}"

                return new_url

Note that the implementation of get_base_url_for_service will depend on how you are storing (env variables, hardcoded, etc.) the host names, ports, and any path prefixes.

Running the Server

We run our proxy server from Docker. Add the following simple Dockerfile:

FROM python:3.8-slim-buster

COPY ./requirements.txt /proxy/requirements.txt
RUN pip install -r /proxy/requirements.txt

COPY ./api_proxy.py /proxy/api_proxy.py

WORKDIR /proxy
CMD ["python3", "-u", "api_proxy.py"]

With that in place, we can add a new “proxy” service to our docker-compose.yaml, making sure to include each backend service as dependencies.

services:
  proxy:
    build:
      context: .
      dockerfile: Dockerfile
    depends_on:
      - users
      - organizations

Conclusion

Using a proxy server like this has helped our team confine request forwarding and transformation into a single place so that our API clients only have to worry about a single base URL. I hope this simple Python proxy server can act as a good starting point for wiring up some of the functionality that would ordinarily be provided through an API Gateway such as Azure’s API Management.