Return the UserOrganization row, or None. Does NOT check superadmin.
(user_id: int, org_id: int, db_session: AsyncSession)
| 20 | |
| 21 | |
| 22 | async def get_user_org(user_id: int, org_id: int, db_session: AsyncSession) -> Optional[UserOrganization]: |
| 23 | """Return the UserOrganization row, or None. Does NOT check superadmin.""" |
| 24 | _cache = getattr(db_session, '_user_org_cache', None) |
| 25 | if _cache is None: |
| 26 | db_session._user_org_cache = {} |
| 27 | _cache = db_session._user_org_cache |
| 28 | key = (user_id, org_id) |
| 29 | if key in _cache: |
| 30 | return _cache[key] |
| 31 | statement = select(UserOrganization).where( |
| 32 | UserOrganization.user_id == user_id, |
| 33 | UserOrganization.org_id == org_id, |
| 34 | ) |
| 35 | result = (await db_session.execute(statement)).scalars().first() |
| 36 | _cache[key] = result |
| 37 | return result |
| 38 | |
| 39 | |
| 40 | async def is_org_member(user_id: int, org_id: int, db_session: AsyncSession) -> bool: |