Resolve a user_id into a fully-populated :class:`AnalyticsContext`. Performs user lookup via the configured AnalyticsUserProvider, extracts consent and org_id, and wraps everything in try/except so no exception ever leaks to the caller. Returns a safe default (consented=False, org_
(user_id: str)
| 64 | |
| 65 | |
| 66 | async def resolve_analytics_context(user_id: str) -> AnalyticsContext: |
| 67 | """Resolve a user_id into a fully-populated :class:`AnalyticsContext`. |
| 68 | |
| 69 | Performs user lookup via the configured AnalyticsUserProvider, extracts |
| 70 | consent and org_id, and wraps everything in try/except so no exception |
| 71 | ever leaks to the caller. |
| 72 | |
| 73 | Returns a safe default (consented=False, org_id=None, user=None) when the |
| 74 | user is not found or any error occurs. |
| 75 | """ |
| 76 | try: |
| 77 | provider = _get_user_provider() |
| 78 | user = await provider.get_user_by_id(user_id) |
| 79 | |
| 80 | if user is None: |
| 81 | return AnalyticsContext(user_id=user_id, **_SAFE_DEFAULT_KWARGS) |
| 82 | |
| 83 | # None = undecided = not consented (same logic as auth.py) |
| 84 | consented = user.user_consents_to_analytics is True |
| 85 | org_id = str(user.current_org_id) if user.current_org_id else None |
| 86 | |
| 87 | return AnalyticsContext( |
| 88 | user_id=user_id, |
| 89 | consented=consented, |
| 90 | org_id=org_id, |
| 91 | user=user, |
| 92 | ) |
| 93 | except Exception: |
| 94 | logger.warning( |
| 95 | 'resolve_analytics_context failed for user_id=%s, returning safe default', |
| 96 | user_id, |
| 97 | ) |
| 98 | return AnalyticsContext(user_id=user_id, **_SAFE_DEFAULT_KWARGS) |