Method validate_api_key

enterprise/storage/api_key_store.py:206–245

Validate an API key and return the associated user_id and org_id if valid. Returns: ApiKeyValidationResult if the key is valid, None otherwise. The org_id may be None for legacy API keys that weren't bound to an organization.

(self, api_key: str)

        return api_key

    async def validate_api_key(self, api_key: str) -> ApiKeyValidationResult | None:
        """Validate an API key and return the associated user_id and org_id if valid.

        Returns:
            ApiKeyValidationResult if the key is valid, None otherwise.
            The org_id may be None for legacy API keys that weren't bound to an organization.
        """
        now = datetime.now(UTC)

        async with a_session_maker() as session:
            result = await session.execute(select(ApiKey).filter(ApiKey.key == api_key))
            key_record = result.scalars().first()

            if not key_record:
                return None

            # expires_at is stored as naive UTC; re-attach tzinfo for comparison.
            if key_record.expires_at:
                expires_at = key_record.expires_at
                if expires_at.tzinfo is None:
                    expires_at = expires_at.replace(tzinfo=UTC)

                if expires_at < now:
                    logger.info(f'API key has expired: {key_record.id}')
                    return None

            # Update last_used_at timestamp
            await session.execute(
                update(ApiKey)
                .where(ApiKey.id == key_record.id)
                .values(last_used_at=now.replace(tzinfo=None))
            )
            await session.commit()

            return ApiKeyValidationResult(
                user_id=key_record.user_id,
                org_id=key_record.org_id,
                key_id=key_record.id,
                key_name=key_record.name,
            )

    async def delete_api_key(self, api_key: str) -> bool:
        """Delete an API key by the key value."""