blob: 9df139661d9ad90e5abf72a07ceec4fa6e83e8e0 [file] [log] [blame]
import logging
from typing import List, Optional, Tuple
from django.http import HttpRequest
from ninja.security.http import HttpAuthBase
from trackingserver_auth.models import Team, User, UserTeamMembership
from trackingserver_base.auth.sync import ensure_user_only_part_of_orgs
logger = logging.getLogger(__name__)
class LocalAPIAuthenticator(HttpAuthBase):
async def __call__(self, request: HttpRequest) -> Optional[Tuple[User, List[Team]]]:
"""This is an authentication client for local mode. It will ensure a user exists with
the appropriate username.
"""
username = request.headers.get("x-api-user")
return await self.ensure(username)
@staticmethod
async def ensure(username: str):
try:
user = await User.objects.aget(email=username)
except User.DoesNotExist:
logger.warning(f"Creating new user {username} for local mode")
user = User(
first_name=username.split("@")[0],
last_name="local account", # TODO -- get the actual last name
email=username,
)
await user.asave()
public = await Team.objects.aget(name="Public")
await ensure_user_only_part_of_orgs(user, [public])
return user, [public]
teams = [
item.team
async for item in UserTeamMembership.objects.filter(user=user)
.all()
.prefetch_related("team")
]
return user, teams