Need help tracking API usage

[ariminia]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from typing import Callable
from fastapi import Request, Response, Depends
from fastapi.routing import APIRoute
from server.routes.users import fastapi_users
from server.models.users import *

class session_log(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request,
        user: User = fastapi_users.get_current_user
        ) -> Response:
            response: Response = await original_route_handler(request)
            # print(f"route response: {response}")
            try:
                print("User", str(user))
            except Exception as e:
                print(f"Error : {e}")
            return response

        return custom_route_handler

Description

Hi, I'm looking for a way to track api usage on user basis. We are using fastapi-users for user management. Can anyone please suggest a way to do this. I tried using a custom API Router but was unable to fetch user details as it done through dependency injection and I kept getting errors like Depends object has no attribute ID/ Depends object is not subscriptable.

Any help would be greatly appreciated.

Thank you!!

Operating System

Linux

Operating System Details

No response

FastAPI Version

fastapi==0.61.1

Python Version

3.8

Additional Context

No response

[AlrasheedA]

Hi @ari-jain013, if your logs don't require the response object, you could create a dependency for logging (that depends on user dependency) and inject it on all routes.

For logging, check: #778 (comment)

For adding the dependency on all paths, check: https://fastapi.tiangolo.com/tutorial/bigger-applications/

And we can add a list of dependencies that will be added to all the path operations in the router and will be executed/solved for each request made to them.

EDIT: Just to add an example

from fastapi import FastAPI, APIRouter, Depends

app = FastAPI()

def user_dep() -> int:
    print("CALLED USER DEPENDENCY")
    user = 0
    return user

def logger_dep(user: int = Depends(user_dep)):
    print('CALLED LOGGER', user)

logged_router = APIRouter(dependencies=[Depends(logger_dep)])

@app.get("/not-logged")
async def not_logged(user: int = Depends(user_dep)):
    return user

@logged_router.get("/logged")
async def logged(user: int = Depends(user_dep)):
    return user

app.include_router(logged_router)

You can confirm, user dependency is only called once for each path (other calls use cached value):

from fastapi.testclient import TestClient
client = TestClient(app)
client.get('/logged')

Gives:

CALLED USER DEPENDENCY
CALLED LOGGER 0

[ariminia]

Okay, I think I can start with this but again it would be great analytics if I could also track is the response was successful or not.

Thank you for the quick reply

[AlrasheedA]

If you need to wait for the response, I would add a middleware for logging. For user info, starlette allows you attach states to the request object, which you can then access from the middleware. Not sure if there is a downside to using this with FastAPI, but the following example runs correctly.

from fastapi import FastAPI, Depends, Request

async def user_dep() -> int:
    user = 0
    return user

async def add_request_state(request: Request, user: int = Depends(user_dep)):
    request.state.user = user

app = FastAPI(dependencies=[Depends(add_request_state)])

@app.middleware("http")
async def add_logs(request: Request, call_next):
    response = await call_next(request)
    print ("some logging logic for user", request.state.user)
    print ("response status code: ", response.status_code)
    return response


@app.get("/test")
async def logged(user: int=Depends(user_dep)):
    return user

# Test behavior
from fastapi.testclient import TestClient
client = TestClient(app)
client.get('/test')

some logging logic for user 0
response status code: 200

[ariminia]

Hi Ahmed, I tried adding a middleware as you suggested

from server.routes.users import fastapi_users
from server.models.users import *
from fastapi import FastAPI, Depends, Request
from fastapi_users import FastAPIUsers


async def add_request_state(request: Request, user: User = Depends(fastapi_users)):
    request.state.user = user


app = FastAPI(
    title="Subtl API",
    description="End to End API interface for SubtlAI",
    version="0.1.0",
    docs_url="/",
    root_path=ROOT_PATH,
    dependencies=[Depends(add_request_state)]
    # root_path_in_servers=False
)

@app.middleware("http")
async def add_logs(request: Request, call_next):
    response = await call_next(request)
    try:
        print ("some logging logic for user", request.state.user.id)
        print ("response status code: ", response.status_code)
    except Exception as e:
        print ("error: ", e)
    return response

And got the following error

api_1 | error: 'State' object has no attribute 'user'

[ariminia]

Hi Ahmed, I tried the first option as well

def logger_dep(user: User = Depends(fastapi_users.get_current_user)):
    print('CALLED LOGGER', str(user.id))

router = APIRouter(dependencies=[Depends(logger_dep)])

And got the following error

TypeError: init() got an unexpected keyword argument 'dependencies'

[AlrasheedA]

Hi @ari-jain013, I just realized shared parameters (e.g. dependencies) support only started at release 0.62 and yours is 0.61.

[upeshphable]

You can use a middleware like the one below which gives timing, generates a unique request id and prints status code as well.

current_request_id is a context variable

class RequestTimeMiddleware(BaseHTTPMiddleware):

    async def dispatch(self, request, call_next):

        idem = "".join(random.choices(string.ascii_uppercase + string.digits, k=20))
        current_request_id.set(idem)

        logger.info(f"rid={idem} start request path={request.url.path}")
        start_time = time.time()

        response = await call_next(request)
        response.headers["X-Request-Id"] = idem

        process_time = (time.time() - start_time) * 1000
        formatted_process_time = "{0:.2f}".format(process_time)

        logger.info(
            f"rid={idem} client-ip={request.client.host} completed_in={formatted_process_time}ms status_code={response.status_code}"
        )
        return response

[ariminia]

Hi @ari-jain013, I just realized shared parameters (e.g. dependencies) support only started at release 0.62 and yours is 0.61.

Yes, I realized the same while looking at the second error.

[ariminia]

You can use a middleware like the one below which gives timing, generates a unique request id and prints status code as well.

current_request_id is a context variable

class RequestTimeMiddleware(BaseHTTPMiddleware):

    async def dispatch(self, request, call_next):

        idem = "".join(random.choices(string.ascii_uppercase + string.digits, k=20))
        current_request_id.set(idem)

        logger.info(f"rid={idem} start request path={request.url.path}")
        start_time = time.time()

        response = await call_next(request)
        response.headers["X-Request-Id"] = idem

        process_time = (time.time() - start_time) * 1000
        formatted_process_time = "{0:.2f}".format(process_time)

        logger.info(
            f"rid={idem} client-ip={request.client.host} completed_in={formatted_process_time}ms status_code={response.status_code}"
        )
        return response

Can you please provide an example on how to use the Middleware class?

[upeshphable]

Just add to the list of middlewares of the app as below:

app.add_middleware(RequestTimeMiddleware)

See documentation here: https://fastapi.tiangolo.com/advanced/middleware/

[ariminia]

Hi @ari-jain013, I just realized shared parameters (e.g. dependencies) support only started at release 0.62 and yours is 0.61.

Hey Ahmed, I upgraded my fastapi version and was able to make shared parameter work. Any suggestion on the other option where I can track response along with request?

But either way, thanks a lot for your help with clean, simple and easy to implement examples.

[ariminia]

Just add to the list of middlewares of the app as below:

app.add_middleware(RequestTimeMiddleware)

See documentation here: https://fastapi.tiangolo.com/advanced/middleware/

Gonna try this now! Thanks @upeshphable

[upeshphable]

@ari-jain013 Please close the issue if it worked.

[ariminia]

Hi @upeshphable i tried the custom middleware

class RequestTimeMiddleware(BaseHTTPMiddleware):

    async def dispatch(self, request, call_next, user:User = Depends(fastapi_users.get_current_user)):
        try:
            print("User: ",str(user.id))
        except Exception as e:
            print("Could not find user. Error : ",e)
        response = await call_next(request)

        return response

app.add_middleware(RequestTimeMiddleware)

and still got the same response

Error : 'Depends' object has no attribute 'id'

[upeshphable]

This is not related to the middleware. You need to check what your dependency, fastapi_users.get_current_user yields and then that user needs to have the id field because you are printing it.

How come you have dependency in dispatch arguments? I don't think you can do that.

[ariminia]

@router.post("", response_model=TransactionOut, status_code=HTTP_201_CREATED)
async def transaction(
    transaction: TransactionIn,
    background_tasks: BackgroundTasks,
    user: User = Depends(fastapi_users.get_current_user),
):
    print( f"Query transaction initiated by {user.email}")
    transaction = await add_transaction( user, transaction)
    if not transaction:
        raise HTTPException(status_code=404, detail="Target (document/group) not found")
    return transaction

The dependency yields user object which has id, email and other fields. but it only works on endpoints or other dependencies. I'm trying to find a way to get user info and response status for all endpoints so I can maintain records of the number of times an endpoint was accessed and the response status.

[ariminia]

Hi Ahmed, I tried the first option as well

def logger_dep(user: User = Depends(fastapi_users.get_current_user)):
    print('CALLED LOGGER', str(user.id))

router = APIRouter(dependencies=[Depends(logger_dep)])

And got the following error

TypeError: init() got an unexpected keyword argument 'dependencies'

This method worked after updating fastapi version but as it is a dependency, I was unable to get the response status to log

[upeshphable]

I think you cannot do it in middleware since it's handled in starlette. You might want to set user as a context variable and then use it here in this method to print out.

@router.post("", response_model=TransactionOut, status_code=HTTP_201_CREATED)
async def transaction(
    transaction: TransactionIn,
    background_tasks: BackgroundTasks,
    user: User = Depends(fastapi_users.get_current_user),
):
    print( f"Query transaction initiated by {user.email}")
    transaction = await add_transaction( user, transaction)
    if not transaction:
        raise HTTPException(status_code=404, detail="Target (document/group) not found")
    return transaction

The dependency yields user object which has id, email and other fields. but it only works on endpoints or other dependencies. I'm trying to find a way to get user info and response status for all endpoints so I can maintain records of the number of times an endpoint was accessed and the response status.

I think you cannot do it in middleware since it's handled in starlette. You might want to set user as a context variable and then use it here in this method to print out.

[ariminia]

A similar issue would be
#991
Apparently the user who asked this solved it but the solution is not provided
Also this stack overflow doubt
https://stackoverflow.com/questions/69913998/how-to-get-current-active-user-in-middleware-fastapi-python