Validate an API token and return an APITokenUser if valid. Args: token: The full token string (lh_...) db_session: Database session Returns: APITokenUser if valid, None otherwise
(
token: str,
db_session: AsyncSession,
)
| 486 | |
| 487 | |
| 488 | async def validate_api_token( |
| 489 | token: str, |
| 490 | db_session: AsyncSession, |
| 491 | ) -> Optional[APITokenUser]: |
| 492 | """ |
| 493 | Validate an API token and return an APITokenUser if valid. |
| 494 | |
| 495 | Args: |
| 496 | token: The full token string (lh_...) |
| 497 | db_session: Database session |
| 498 | |
| 499 | Returns: |
| 500 | APITokenUser if valid, None otherwise |
| 501 | """ |
| 502 | from src.services.api_tokens.api_tokens import validate_api_token_for_auth |
| 503 | |
| 504 | # Validate the token using the service |
| 505 | api_token = await validate_api_token_for_auth(token, db_session) |
| 506 | |
| 507 | if not api_token: |
| 508 | return None |
| 509 | |
| 510 | # Normalize rights to a plain dict of plain dicts so .get() works everywhere |
| 511 | raw_rights = api_token.rights |
| 512 | if raw_rights is None: |
| 513 | rights = None |
| 514 | elif isinstance(raw_rights, dict): |
| 515 | # Already a dict, but inner values might be Pydantic models |
| 516 | rights = {} |
| 517 | for k, v in raw_rights.items(): |
| 518 | if isinstance(v, dict): |
| 519 | rights[k] = v |
| 520 | elif hasattr(v, 'model_dump'): |
| 521 | rights[k] = v.model_dump() |
| 522 | elif hasattr(v, 'dict'): |
| 523 | rights[k] = v.dict() |
| 524 | else: |
| 525 | rights[k] = v |
| 526 | else: |
| 527 | # Full Pydantic Rights model |
| 528 | rights = raw_rights.model_dump() if hasattr(raw_rights, 'model_dump') else raw_rights.dict() |
| 529 | |
| 530 | # Create and return an APITokenUser |
| 531 | return APITokenUser( |
| 532 | id=api_token.id, |
| 533 | user_uuid=api_token.token_uuid, |
| 534 | username=f"api_token_{api_token.name}", |
| 535 | org_id=api_token.org_id, |
| 536 | rights=rights, |
| 537 | token_name=api_token.name, |
| 538 | created_by_user_id=api_token.created_by_user_id, |
| 539 | ) |
| 540 | |
| 541 | |
| 542 | async def validate_superadmin_api_token( |