Create a proxy server in python (Flask, Fastapi) that will map to s3 bucket

Question

I have deployed the static site in the aws s3 bucket that's URL looks like this -

https://{bucket_name}.s3.{zone}.amazonaws.com/website/{ID}/index.html

and I want to redirect all the traffic that is fall into the server, should be forward into the s3 bucket like this -

username.localhost:9000 -> https://{bucket_name}.s3.{zone}.amazonaws.com/website/{username}/index.html

if you have any suggestion from nginx please share

I have tried solution -

import uvicorn
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse
import httpx
import boto3

app = FastAPI()
S3_BUCKET_NAME = ${bucket}

async def proxy_request(request: Request):
    s3_client = boto3.client('s3')

    # Construct the S3 object key based on your requirements
    x = request.url.path.split("/")[-2:]
    y = '/'.join(x)
    s3_key = f"website/{y}"

    try:
        # Get the object from S3
        response = s3_client.get_object(Bucket=S3_BUCKET_NAME, Key=s3_key)
    except s3_client.exceptions.NoSuchKey:
        raise HTTPException(status_code=404, detail="Object not found")

    # Return the S3 object's content back to the client
    return StreamingResponse(
        content=response['Body'].iter_chunks(),
        status_code=200,
        headers={"Content-Type": response['ContentType']},
    )

@app.middleware("http")
async def proxy_middleware(request: Request, call_next):
    try:
        return await proxy_request(request)
    except HTTPException as exc:
        return exc

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=9000)

Answer 1

You're asking for solutions that fundamentally differ from each other. Your summary says proxy, but description says redirect. Also your localhost URL has a subdomain, which is usable only if you have it mapped in your host entry file. I'll try to cover the use cases on localhost to s3 redirect/forwarding.

If all you want is a simple redirect,

@app.route('/{_:path}')
async def s3_redirect(request: Request) -> RedirectResponse:
    # return RedirectResponse(request.url.replace())  # If you want to retain the path/query params
    return RedirectResponse("https://{bucket_name}.s3.{zone}.amazonaws.com/website/{username}/index.html")

If you're looking for a solution with nginx, the following config should do the trick

server {
    listen 9000;

    location / {
        proxy_pass {s3_website};
    }
    # more locations to different endpoints
}

I'd prefer nginx or an equivalent proxy application over using FastAPI if the use case is just proxy. However if you still want to use FastAPI, I'd use a somewhat extensive solution like the one below. The following example will act as a basic proxy service. You wouldn't need to check for bucket objects since the httpx client will try to reach the s3 endpoint and respond accordingly.

Note that your bucket objects should be publicly accessible to use this solution.

import logging
from http import HTTPStatus

import httpx
import uvicorn.logging
from fastapi import FastAPI, HTTPException, Request, Response
from fastapi.routing import APIRoute

TARGET_URL = "http://127.0.0.1:8080"  # Change this to the URL you want to proxy requests to


async def proxy(request: Request) -> Response:
    """Proxy handler function to forward incoming requests to a target URL.

    Args:
        request: The incoming request object.

    Returns:
        Response: The response object with the forwarded content and headers.
    """
    try:
        async with httpx.AsyncClient() as client:
            body = await request.body()
            # noinspection PyTypeChecker
            response = await client.request(
                method=request.method,
                url=TARGET_URL + request.url.path,
                headers=dict(request.headers),
                cookies=request.cookies,
                params=dict(request.query_params),
                data=body.decode(),
            )

            # If the response content-type is text/html, we need to rewrite links in it
            content_type = response.headers.get("content-type", "")
            if "text/html" in content_type:
                content = response.text
                # Modify content if necessary (e.g., rewriting links)
                # content = modify_html_links(content)
            else:
                content = response.content
            response.headers.pop("content-encoding", None)

            return Response(content, response.status_code, response.headers, content_type)
    except httpx.RequestError as exc:
        logging.getLogger("uvicorn.error").error(exc)
        raise HTTPException(status_code=HTTPStatus.BAD_GATEWAY.value, detail=HTTPStatus.BAD_GATEWAY.phrase)


if __name__ == '__main__':
    uvicorn.run(
        app=FastAPI(
            routes=[
                APIRoute("/{_:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"], endpoint=proxy)
            ]
        )
    )