Access check for a datasource by account and container using roles from a JWT claim. Parameters ---------- account : str The account name. container : str The container name. roles : List[str] The roles from the JWT claim. Returns -------
(account: str, container: str, user=Depends(get_current_user))
| 4 | |
| 5 | |
| 6 | async def check_access(account: str, container: str, user=Depends(get_current_user)) -> str | None: |
| 7 | """ |
| 8 | Access check for a datasource by account and container using roles from a JWT claim. |
| 9 | |
| 10 | Parameters |
| 11 | ---------- |
| 12 | account : str |
| 13 | The account name. |
| 14 | container : str |
| 15 | The container name. |
| 16 | roles : List[str] |
| 17 | The roles from the JWT claim. |
| 18 | |
| 19 | Returns |
| 20 | ------- |
| 21 | "public", "reader", "owner", or None |
| 22 | """ |
| 23 | groups = user.get("groups", []) |
| 24 | if isinstance(groups, str): |
| 25 | groups = [groups] |
| 26 | groups.append(user.get("preferred_username")) |
| 27 | |
| 28 | data_source = await db().datasources.find_one( |
| 29 | {"account": account, "container": container} |
| 30 | ) |
| 31 | |
| 32 | if data_source: |
| 33 | if 'owners' in data_source and any(group in data_source['owners'] for group in groups): |
| 34 | return "owner" |
| 35 | if 'readers' in data_source and any(group in data_source['readers'] for group in groups): |
| 36 | return "reader" |
| 37 | if 'public' in data_source and data_source['public']: |
| 38 | return "public" |
| 39 | return None |
no test coverage detected