Async Multi-Tenant Scope Checking on Views

[tonydattolo]

Using django 4.2 and django-ninja 1.1.0

I'm using django-scopes to handle multi-tenancy tenant scoping permissions for our views. I have implemented an async version of the two scope functions as well in an attempt to get it to work. I'm working if anyone has suggestions or has dealt with similar tenant scope checks in views.

works fine:

def list_agents(request):
    try:
        with scope(workspace=request.COOKIES.get("workspace")):
            agents = Agent.objects.all()
            return agents
    except Exception as e:
        raise HttpError(status=500, message=str(e))

throws "detail": "Unauthorized"

async def list_agents(request):
    try:
        async with scope(workspace=request.COOKIES.get("workspace")):
            # agents = await sync_to_async(list)(Agent.objects.all())
            agents = await Agent.objects.afirst()
            return agents
    except Exception as e:
        raise HttpError(status=500, message=str(e))

our jwt httponly cookie ninja auth:

from rest_framework_simplejwt.authentication import JWTAuthentication
from ninja.security import APIKeyCookie
class JWTAuthenticationCookie(APIKeyCookie):
    param_name = "access"  # Name of the cookie containing the JWT token

    def authenticate(self, request, token: str) -> Optional[Any]:
        jwt_authentication = JWTAuthentication()

        try:
            print(f"Token received: {token}")
            if token:
                validated_token = jwt_authentication.get_validated_token(token)
                user = jwt_authentication.get_user(validated_token)
                if user is not None:
                    return user
        except Exception as e:
            pass

        return None

async scopes attempt (regular django_scopes doesn't work, maintainer says it should so not sure, maybe ninja auth issues?):

from django_scopes import scope, scopes_disabled

from contextlib import asynccontextmanager
from contextvars import ContextVar
from typing import Optional

state: ContextVar[Optional[dict]] = ContextVar("state", default=None)


@asynccontextmanager
async def async_scope(**scope_kwargs):
    previous_scope = state.get()
    if previous_scope is None:
        previous_scope = {}
        state.set(previous_scope)

    new_scope = dict(previous_scope)
    new_scope["_enabled"] = True
    new_scope.update(scope_kwargs)
    state.set(new_scope)
    try:
        yield
    finally:
        state.set(previous_scope)


@asynccontextmanager
async def async_scopes_disabled():
    async with scope(_enabled=False):
        yield

auth enabled globally:

api = NinjaAPI(
    openapi_extra={
        "info": {
            "termsOfService": "https://example.com/terms/",
        }
    },
    title="Demo API",
    description="This is a demo API with dynamic OpenAPI info section",
    version="2.0.0",
    auth=[JWTAuthenticationCookie()],
    csrf=False,
    docs=Swagger(
        settings={
            "persistAuthorizationInCookie": "access",
            "cookieSecurity": [{"access": []}],
            "persistAuthorization": True,
        }
    ),
)