0

I wrote my own session logic and use the following decorator to check the request:

def require_session(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        request = kwargs['request']
        session_key = request.headers.get('X-Session-Key')
        if session_key is not None:
            session = Session.check(session_key)  # Check in db
            if session is not None:
                return await func(*args, **kwargs)
        raise exc.access_denied()
    return wrapper

And I use it like this

import fastapi as fa

employees_router = fa.APIRouter(default_response_class=fa.responses.JSONResponse)

@employees_router.get('/{id}')
@require_session
async def get_employee(request: fa.Request, id: int) -> EmployeeDBSchema:
    employee = Employee.get_or_none(id=id)
    if employee is None:
        raise exc.instance_by_field_not_found(Employee, id)
    return employee

I simply check for the session key first in the headers and then in the database. And everything works fine until I try to call these methods inside other methods.

@administrators_router.post('')
@require_session
async def create_administrator(request: fa.Request, data: AdministratorSchema) -> AdministratorDBSchema:
    data_dump = data.model_dump()
    if isinstance(data_dump['employee'], int):
        await get_employee(request, data_dump['employee'])  # Inner call
    try:
        admin = Administrator.create(**data_dump)
    except pw.IntegrityError as e:
        raise exc.integrity(e)
    return admin

This results in the session being checked twice. How can I avoid the second session check?

I was looking for something like "Parameter that I can set only inside the fastapi app" to just define an additional parameter inner_call=false that tells the decorator not to check the session, but I didn't find anything.

The second solution is to separate the endpoint from its implementation like this:

@employees_router.get('/{id}')
@require_session
async def get_employee(request: fa.Request, id: int) -> EmployeeDBSchema:
    return get_employee_impl(id)

async def get_employee_impl(id: int) -> EmployeeDBSchema:
    employee = Employee.get_or_none(id=id)
    if employee is None:
        raise exc.instance_by_field_not_found(Employee, id)
    return employee

And just call get_employee_impl where I want to avoid the session check, but rewriting all the functions can take a lot of time. Also, this solution doesn't seem very clean to me.

1 Answer 1

0

Use contextvar, which can store variable in a request context

modify your code like:

from contextvars import ContextVar
ctx_session = ContextVar('ctx_session', default=None)

def require_session(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        if ctx_session.get() is None:
            request = kwargs['request']
            session_key = request.headers.get('X-Session-Key')
            if session_key is not None:
                session = Session.check(session_key)  # Check in db
                if session is not None:
                    ctx_session.set(session)
                    return await func(*args, **kwargs)
            raise exc.access_denied()
        else:
            return await func(*args, **kwargs)
    return wrapper
Sign up to request clarification or add additional context in comments.

Comments

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.