Modifying path param on the fly in FastAPI

Question

We have a Lambda-based FastAPI application that has a bunch of routers such as this

class ReportsRouter:
    @router.post(
        "/{customer_id}/report", response_model=ReportResponse,
    )
    async def report(customer_id: int, request: ReportRequest):
        return await ReportGenerator(customer_id, request)

For years, we have been supporting only one tenant (not 1 customer, but 1 tenant providing us hundreds of customers). So, we have our customer_ids starting from 1, 2 and so on.

Now, we have a new tenant integrating with us and they have their own ecosystem and databases and their customer_ids also start from 1. We want to be able to extend our Lambda service to this new tenant but without a lot of code changes. But the colliding customer_id is a big blocker for us, since we need to understand if an API call is coming for customer_id 1, is it for the older tenant or the newer one?

How we decided to solve this colliding customer_id issue is by adding an offset. We know that our current tenant has only 1000 customers and even after 5 years, they probably won't give us more than 500. This means we can safely add an offset of 5000 to the customer_ids coming from our new tenant.

For ex, our new tenant gives us data for customer_id 63, but we internally convert that into 5063 and store the same in our DB.

While this idea theoretically works for us, we faced a problem with modifying the path params on the fly. We wrote a middleware like thus (following this FastAPI Github issue)

@app.middleware("http")
async def offset_customer_id(request, call_next):
    tenant_id = request.headers.get("tenant_id")
    
    if tenant_id == "new_tenant":  # just an example
        routes = request.app.router.routes
        for route in routes:
            match, scope = route.matches(request)
            if match == Match.FULL:
                customer_id = scope["path_params"].get("customer_id")
                if customer_id:
                    customer_id = int(customer_id) + 5000
                    scope["path_params"]["customer_id"] = str(customer_id)

    response = await call_next(request)
    return response

But of course, it doesn't work, because the path param is once again calculated fresh by Starlette when call_next is invoked, and the same value is passed to the router.

The only other way we could think of is to put this offsetting logic in every single router we have, which is obviously a big effort. Something like this

class ReportRequest:
   tenant_id: int = Field()  # add tenant_id into this DTO first

class ReportsRouter:
    @router.post(
        "/{customer_id}/report", response_model=ReportResponse,
    )
    async def report(customer_id: int, request: ReportRequest):
        if request.tenant_id == "new_tenant":
            customer_id += 500
        return await ReportGenerator(customer_id, request)

Don't think FastAPI routers have "base classes" (at least nothing seen from the official docs or Google results), so it seems like we have to replicate that if clause and offsetting operation in every router.

So, we wanted to know if there's any cleaner way of achieving this. Any help would be appreciated. Thanks!

Answer 1

One idea could be to use a dependency for offsetting the ID.

def customer_id_with_offset(customer_id: int, request: Request) -> int:
    tenant_id = request.headers.get("tenant_id")
    
    if tenant_id == "new_tenant":
        customer_id += 5000

    return customer_id

Then in the routes change the type hint as follows:

async def report(request: ReportRequest, customer_id: int = Depends(customer_id_with_offset)):

or for a more modern FastAPI notation:

async def report(customer_id: Annotated[int, Depends(customer_id_with_offset)], request: ReportRequest):

Still need to refactor all the routes, though