fief_client.integrations.fastapi

FastAPI integration.

View Source

  1"""FastAPI integration."""
  2import uuid
  3from inspect import Parameter, Signature, isawaitable
  4from typing import (
  5    AsyncGenerator,
  6    Callable,
  7    Coroutine,
  8    Generator,
  9    List,
 10    Optional,
 11    Protocol,
 12    TypeVar,
 13    Union,
 14    cast,
 15)
 16
 17from fastapi import Depends, HTTPException, Request, Response, status
 18from fastapi.security.base import SecurityBase
 19from fastapi.security.http import HTTPAuthorizationCredentials
 20from makefun import with_signature
 21
 22from fief_client import (
 23    Fief,
 24    FiefAccessTokenACRTooLow,
 25    FiefAccessTokenExpired,
 26    FiefAccessTokenInfo,
 27    FiefAccessTokenInvalid,
 28    FiefAccessTokenMissingPermission,
 29    FiefAccessTokenMissingScope,
 30    FiefACR,
 31    FiefAsync,
 32    FiefUserInfo,
 33)
 34
 35FiefClientClass = Union[Fief, FiefAsync]
 36
 37TokenType = Union[str, HTTPAuthorizationCredentials]
 38
 39RETURN_TYPE = TypeVar("RETURN_TYPE")
 40
 41DependencyCallable = Callable[
 42    ...,
 43    Union[
 44        RETURN_TYPE,
 45        Coroutine[None, None, RETURN_TYPE],
 46        AsyncGenerator[RETURN_TYPE, None],
 47        Generator[RETURN_TYPE, None, None],
 48    ],
 49]
 50
 51
 52class UserInfoCacheProtocol(Protocol):
 53    """
 54    Protocol that should follow a class to implement a cache mechanism for user information.
 55
 56    Read more: https://docs.fief.dev/integrate/python/fastapi/#caching-user-information
 57    """
 58
 59    async def get(self, user_id: uuid.UUID) -> Optional[FiefUserInfo]:
 60        """
 61        Retrieve user information from cache, if available.
 62
 63        :param user_id: The ID of the user to retrieve information for.
 64        """
 65        ...  # pragma: no cover
 66
 67    async def set(self, user_id: uuid.UUID, userinfo: FiefUserInfo) -> None:
 68        """
 69        Store user information in cache.
 70
 71        :param user_id: The ID of the user to cache information for.
 72        :param userinfo: The user information to cache.
 73        """
 74        ...  # pragma: no cover
 75
 76
 77class FiefAuth:
 78    """
 79    Helper class to integrate Fief authentication with FastAPI.
 80
 81    **Example:**
 82
 83    ```py
 84    from fastapi.security import OAuth2AuthorizationCodeBearer
 85    from fief_client import FiefAccessTokenInfo, FiefAsync
 86    from fief_client.integrations.fastapi import FiefAuth
 87
 88    fief = FiefAsync(
 89        "https://example.fief.dev",
 90        "YOUR_CLIENT_ID",
 91        "YOUR_CLIENT_SECRET",
 92    )
 93
 94    scheme = OAuth2AuthorizationCodeBearer(
 95        "https://example.fief.dev/authorize",
 96        "https://example.fief.dev/api/token",
 97        scopes={"openid": "openid", "offline_access": "offline_access"},
 98    )
 99
100    auth = FiefAuth(fief, scheme)
101    ```
102    """
103
104    def __init__(
105        self,
106        client: FiefClientClass,
107        scheme: SecurityBase,
108        *,
109        get_userinfo_cache: Optional[DependencyCallable[UserInfoCacheProtocol]] = None,
110    ) -> None:
111        """
112        :param client: Instance of a Fief client.
113        Can be either `fief_client.Fief` or `fief_client.FiefAsync`.
114        :param scheme: FastAPI security scheme.
115        It'll be used to retrieve the access token in the request.
116        :param get_userinfo_cache: Optional dependency returning an instance of a class
117        following the `UserInfoCacheProtocol`.
118        It'll be used to cache user information on your server.
119        Otherwise, the Fief API will always be reached when requesting user information.
120        """
121        self.client = client
122        self.scheme = scheme
123        self.get_userinfo_cache = get_userinfo_cache
124
125    def authenticated(
126        self,
127        optional: bool = False,
128        scope: Optional[List[str]] = None,
129        acr: Optional[FiefACR] = None,
130        permissions: Optional[List[str]] = None,
131    ):
132        """
133        Return a FastAPI dependency to check if a request is authenticated.
134
135        If the request is authenticated, the dependency will return a `fief_client.FiefAccessTokenInfo`.
136
137        :param optional: If `False` and the request is not authenticated,
138        an unauthorized response will be raised.
139        :param scope: Optional list of scopes required.
140        If the access token lacks one of the required scope, a forbidden response will be raised.
141        :param acr: Optional minimum ACR level required.
142        If the access token doesn't meet the minimum level, a forbidden response will be raised.
143        Read more: https://docs.fief.dev/going-further/acr/
144        :param permissions: Optional list of permissions required.
145        If the access token lacks one of the required permission, a forbidden response will be raised.
146
147        **Example**
148
149        ```py
150        @app.get("/authenticated")
151        async def get_authenticated(
152            access_token_info: FiefAccessTokenInfo = Depends(auth.authenticated()),
153        ):
154            return access_token_info
155        ```
156        """
157        signature = self._get_authenticated_call_signature(self.scheme)
158
159        @with_signature(signature)
160        async def _authenticated(
161            request: Request, response: Response, token: Optional[TokenType]
162        ) -> Optional[FiefAccessTokenInfo]:
163            if token is None:
164                if optional:
165                    return None
166                return await self.get_unauthorized_response(request, response)
167
168            if isinstance(token, HTTPAuthorizationCredentials):
169                token = token.credentials
170
171            try:
172                result = self.client.validate_access_token(
173                    token,
174                    required_scope=scope,
175                    required_acr=acr,
176                    required_permissions=permissions,
177                )
178                if isawaitable(result):
179                    info = await result
180                else:
181                    info = result
182            except (FiefAccessTokenInvalid, FiefAccessTokenExpired):
183                if optional:
184                    return None
185                return await self.get_unauthorized_response(request, response)
186            except (
187                FiefAccessTokenMissingScope,
188                FiefAccessTokenACRTooLow,
189                FiefAccessTokenMissingPermission,
190            ):
191                return await self.get_forbidden_response(request, response)
192
193            return info
194
195        return _authenticated
196
197    def current_user(
198        self,
199        optional: bool = False,
200        scope: Optional[List[str]] = None,
201        acr: Optional[FiefACR] = None,
202        permissions: Optional[List[str]] = None,
203        refresh: bool = False,
204    ):
205        """
206        Return a FastAPI dependency to check if a user is authenticated.
207
208        If the request is authenticated, the dependency will return a `fief_client.FiefUserInfo`.
209
210        If provided, the cache mechanism will be used to retrieve this information without calling the Fief API.
211
212        :param optional: If `False` and the request is not authenticated,
213        an unauthorized response will be raised.
214        :param scope: Optional list of scopes required.
215        If the access token lacks one of the required scope, a forbidden response will be raised.
216        :param acr: Optional minimum ACR level required.
217        If the access token doesn't meet the minimum level, a forbidden response will be raised.
218        Read more: https://docs.fief.dev/going-further/acr/
219        :param permissions: Optional list of permissions required.
220        If the access token lacks one of the required permission, a forbidden response will be raised.
221        :param refresh: If `True`, the user information will be refreshed from the Fief API.
222        Otherwise, the cache will be used.
223
224        **Example**
225
226        ```py
227        @app.get("/current-user", name="current_user")
228        async def get_current_user(
229            user: FiefUserInfo = Depends(auth.current_user()),
230        ):
231            return {"email": user["email"]}
232        ```
233        """
234        signature = self._get_current_user_call_signature(
235            self.authenticated(optional, scope, acr, permissions)
236        )
237
238        @with_signature(signature)
239        async def _current_user(
240            access_token_info: Optional[FiefAccessTokenInfo], *args, **kwargs
241        ) -> Optional[FiefUserInfo]:
242            userinfo_cache: Optional[UserInfoCacheProtocol] = kwargs.get(
243                "userinfo_cache"
244            )
245
246            if access_token_info is None and optional:
247                return None
248            assert access_token_info is not None
249
250            userinfo = None
251            if userinfo_cache is not None:
252                userinfo = await userinfo_cache.get(access_token_info["id"])
253
254            if userinfo is None or refresh:
255                result = self.client.userinfo(access_token_info["access_token"])
256                if isawaitable(result):
257                    userinfo = cast(FiefUserInfo, await result)
258                else:
259                    userinfo = cast(FiefUserInfo, result)
260
261                if userinfo_cache is not None:
262                    await userinfo_cache.set(access_token_info["id"], userinfo)
263
264            return userinfo
265
266        return _current_user
267
268    async def get_unauthorized_response(self, request: Request, response: Response):
269        """
270        Raise an `fastapi.HTTPException` with the status code 401.
271
272        This method is called when using the `authenticated` or `current_user` dependency
273        but the request is not authenticated.
274
275        You can override this method to customize the behavior in this case.
276        """
277        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
278
279    async def get_forbidden_response(self, request: Request, response: Response):
280        """
281        Raise an `fastapi.HTTPException` with the status code 403.
282
283        This method is called when using the `authenticated` or `current_user` dependency
284        but the access token doesn't match the list of scopes or permissions.
285
286        You can override this method to customize the behavior in this case.
287        """
288        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
289
290    def _get_authenticated_call_signature(self, scheme: SecurityBase) -> Signature:
291        """
292        Generate a dynamic signature for the authenticated dependency.
293        Here comes some blood magic 🧙‍♂️
294        Thank to "makefun", we are able to generate callable
295        with a dynamic security scheme dependency at runtime.
296        This way, it's detected by the OpenAPI generator.
297        """
298        parameters: List[Parameter] = [
299            Parameter(
300                name="request",
301                kind=Parameter.POSITIONAL_OR_KEYWORD,
302                annotation=Request,
303            ),
304            Parameter(
305                name="response",
306                kind=Parameter.POSITIONAL_OR_KEYWORD,
307                annotation=Response,
308            ),
309            Parameter(
310                name="token",
311                kind=Parameter.POSITIONAL_OR_KEYWORD,
312                default=Depends(cast(Callable, scheme)),
313            ),
314        ]
315
316        return Signature(parameters)
317
318    def _get_current_user_call_signature(self, authenticated: Callable) -> Signature:
319        """
320        Generate a dynamic signature for the current_user dependency.
321        Here comes some blood magic 🧙‍♂️
322        Thank to "makefun", we are able to generate callable
323        with a dynamic security scheme dependency at runtime.
324        This way, it's detected by the OpenAPI generator.
325        """
326        parameters: List[Parameter] = [
327            Parameter(
328                name="access_token_info",
329                kind=Parameter.POSITIONAL_OR_KEYWORD,
330                default=Depends(authenticated),
331                annotation=FiefAccessTokenInfo,
332            ),
333        ]
334
335        if self.get_userinfo_cache is not None:
336            parameters.append(
337                Parameter(
338                    name="userinfo_cache",
339                    kind=Parameter.POSITIONAL_OR_KEYWORD,
340                    default=Depends(self.get_userinfo_cache),
341                    annotation=UserInfoCacheProtocol,
342                ),
343            )
344
345        return Signature(parameters)
346
347
348__all__ = ["FiefAuth", "FiefClientClass", "UserInfoCacheProtocol"]

class FiefAuth: View Source

 78class FiefAuth:
 79    """
 80    Helper class to integrate Fief authentication with FastAPI.
 81
 82    **Example:**
 83
 84    ```py
 85    from fastapi.security import OAuth2AuthorizationCodeBearer
 86    from fief_client import FiefAccessTokenInfo, FiefAsync
 87    from fief_client.integrations.fastapi import FiefAuth
 88
 89    fief = FiefAsync(
 90        "https://example.fief.dev",
 91        "YOUR_CLIENT_ID",
 92        "YOUR_CLIENT_SECRET",
 93    )
 94
 95    scheme = OAuth2AuthorizationCodeBearer(
 96        "https://example.fief.dev/authorize",
 97        "https://example.fief.dev/api/token",
 98        scopes={"openid": "openid", "offline_access": "offline_access"},
 99    )
100
101    auth = FiefAuth(fief, scheme)
102    ```
103    """
104
105    def __init__(
106        self,
107        client: FiefClientClass,
108        scheme: SecurityBase,
109        *,
110        get_userinfo_cache: Optional[DependencyCallable[UserInfoCacheProtocol]] = None,
111    ) -> None:
112        """
113        :param client: Instance of a Fief client.
114        Can be either `fief_client.Fief` or `fief_client.FiefAsync`.
115        :param scheme: FastAPI security scheme.
116        It'll be used to retrieve the access token in the request.
117        :param get_userinfo_cache: Optional dependency returning an instance of a class
118        following the `UserInfoCacheProtocol`.
119        It'll be used to cache user information on your server.
120        Otherwise, the Fief API will always be reached when requesting user information.
121        """
122        self.client = client
123        self.scheme = scheme
124        self.get_userinfo_cache = get_userinfo_cache
125
126    def authenticated(
127        self,
128        optional: bool = False,
129        scope: Optional[List[str]] = None,
130        acr: Optional[FiefACR] = None,
131        permissions: Optional[List[str]] = None,
132    ):
133        """
134        Return a FastAPI dependency to check if a request is authenticated.
135
136        If the request is authenticated, the dependency will return a `fief_client.FiefAccessTokenInfo`.
137
138        :param optional: If `False` and the request is not authenticated,
139        an unauthorized response will be raised.
140        :param scope: Optional list of scopes required.
141        If the access token lacks one of the required scope, a forbidden response will be raised.
142        :param acr: Optional minimum ACR level required.
143        If the access token doesn't meet the minimum level, a forbidden response will be raised.
144        Read more: https://docs.fief.dev/going-further/acr/
145        :param permissions: Optional list of permissions required.
146        If the access token lacks one of the required permission, a forbidden response will be raised.
147
148        **Example**
149
150        ```py
151        @app.get("/authenticated")
152        async def get_authenticated(
153            access_token_info: FiefAccessTokenInfo = Depends(auth.authenticated()),
154        ):
155            return access_token_info
156        ```
157        """
158        signature = self._get_authenticated_call_signature(self.scheme)
159
160        @with_signature(signature)
161        async def _authenticated(
162            request: Request, response: Response, token: Optional[TokenType]
163        ) -> Optional[FiefAccessTokenInfo]:
164            if token is None:
165                if optional:
166                    return None
167                return await self.get_unauthorized_response(request, response)
168
169            if isinstance(token, HTTPAuthorizationCredentials):
170                token = token.credentials
171
172            try:
173                result = self.client.validate_access_token(
174                    token,
175                    required_scope=scope,
176                    required_acr=acr,
177                    required_permissions=permissions,
178                )
179                if isawaitable(result):
180                    info = await result
181                else:
182                    info = result
183            except (FiefAccessTokenInvalid, FiefAccessTokenExpired):
184                if optional:
185                    return None
186                return await self.get_unauthorized_response(request, response)
187            except (
188                FiefAccessTokenMissingScope,
189                FiefAccessTokenACRTooLow,
190                FiefAccessTokenMissingPermission,
191            ):
192                return await self.get_forbidden_response(request, response)
193
194            return info
195
196        return _authenticated
197
198    def current_user(
199        self,
200        optional: bool = False,
201        scope: Optional[List[str]] = None,
202        acr: Optional[FiefACR] = None,
203        permissions: Optional[List[str]] = None,
204        refresh: bool = False,
205    ):
206        """
207        Return a FastAPI dependency to check if a user is authenticated.
208
209        If the request is authenticated, the dependency will return a `fief_client.FiefUserInfo`.
210
211        If provided, the cache mechanism will be used to retrieve this information without calling the Fief API.
212
213        :param optional: If `False` and the request is not authenticated,
214        an unauthorized response will be raised.
215        :param scope: Optional list of scopes required.
216        If the access token lacks one of the required scope, a forbidden response will be raised.
217        :param acr: Optional minimum ACR level required.
218        If the access token doesn't meet the minimum level, a forbidden response will be raised.
219        Read more: https://docs.fief.dev/going-further/acr/
220        :param permissions: Optional list of permissions required.
221        If the access token lacks one of the required permission, a forbidden response will be raised.
222        :param refresh: If `True`, the user information will be refreshed from the Fief API.
223        Otherwise, the cache will be used.
224
225        **Example**
226
227        ```py
228        @app.get("/current-user", name="current_user")
229        async def get_current_user(
230            user: FiefUserInfo = Depends(auth.current_user()),
231        ):
232            return {"email": user["email"]}
233        ```
234        """
235        signature = self._get_current_user_call_signature(
236            self.authenticated(optional, scope, acr, permissions)
237        )
238
239        @with_signature(signature)
240        async def _current_user(
241            access_token_info: Optional[FiefAccessTokenInfo], *args, **kwargs
242        ) -> Optional[FiefUserInfo]:
243            userinfo_cache: Optional[UserInfoCacheProtocol] = kwargs.get(
244                "userinfo_cache"
245            )
246
247            if access_token_info is None and optional:
248                return None
249            assert access_token_info is not None
250
251            userinfo = None
252            if userinfo_cache is not None:
253                userinfo = await userinfo_cache.get(access_token_info["id"])
254
255            if userinfo is None or refresh:
256                result = self.client.userinfo(access_token_info["access_token"])
257                if isawaitable(result):
258                    userinfo = cast(FiefUserInfo, await result)
259                else:
260                    userinfo = cast(FiefUserInfo, result)
261
262                if userinfo_cache is not None:
263                    await userinfo_cache.set(access_token_info["id"], userinfo)
264
265            return userinfo
266
267        return _current_user
268
269    async def get_unauthorized_response(self, request: Request, response: Response):
270        """
271        Raise an `fastapi.HTTPException` with the status code 401.
272
273        This method is called when using the `authenticated` or `current_user` dependency
274        but the request is not authenticated.
275
276        You can override this method to customize the behavior in this case.
277        """
278        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
279
280    async def get_forbidden_response(self, request: Request, response: Response):
281        """
282        Raise an `fastapi.HTTPException` with the status code 403.
283
284        This method is called when using the `authenticated` or `current_user` dependency
285        but the access token doesn't match the list of scopes or permissions.
286
287        You can override this method to customize the behavior in this case.
288        """
289        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
290
291    def _get_authenticated_call_signature(self, scheme: SecurityBase) -> Signature:
292        """
293        Generate a dynamic signature for the authenticated dependency.
294        Here comes some blood magic 🧙‍♂️
295        Thank to "makefun", we are able to generate callable
296        with a dynamic security scheme dependency at runtime.
297        This way, it's detected by the OpenAPI generator.
298        """
299        parameters: List[Parameter] = [
300            Parameter(
301                name="request",
302                kind=Parameter.POSITIONAL_OR_KEYWORD,
303                annotation=Request,
304            ),
305            Parameter(
306                name="response",
307                kind=Parameter.POSITIONAL_OR_KEYWORD,
308                annotation=Response,
309            ),
310            Parameter(
311                name="token",
312                kind=Parameter.POSITIONAL_OR_KEYWORD,
313                default=Depends(cast(Callable, scheme)),
314            ),
315        ]
316
317        return Signature(parameters)
318
319    def _get_current_user_call_signature(self, authenticated: Callable) -> Signature:
320        """
321        Generate a dynamic signature for the current_user dependency.
322        Here comes some blood magic 🧙‍♂️
323        Thank to "makefun", we are able to generate callable
324        with a dynamic security scheme dependency at runtime.
325        This way, it's detected by the OpenAPI generator.
326        """
327        parameters: List[Parameter] = [
328            Parameter(
329                name="access_token_info",
330                kind=Parameter.POSITIONAL_OR_KEYWORD,
331                default=Depends(authenticated),
332                annotation=FiefAccessTokenInfo,
333            ),
334        ]
335
336        if self.get_userinfo_cache is not None:
337            parameters.append(
338                Parameter(
339                    name="userinfo_cache",
340                    kind=Parameter.POSITIONAL_OR_KEYWORD,
341                    default=Depends(self.get_userinfo_cache),
342                    annotation=UserInfoCacheProtocol,
343                ),
344            )
345
346        return Signature(parameters)

Helper class to integrate Fief authentication with FastAPI.

Example:

from fastapi.security import OAuth2AuthorizationCodeBearer
from fief_client import FiefAccessTokenInfo, FiefAsync
from fief_client.integrations.fastapi import FiefAuth

fief = FiefAsync(
    "https://example.fief.dev",
    "YOUR_CLIENT_ID",
    "YOUR_CLIENT_SECRET",
)

scheme = OAuth2AuthorizationCodeBearer(
    "https://example.fief.dev/authorize",
    "https://example.fief.dev/api/token",
    scopes={"openid": "openid", "offline_access": "offline_access"},
)

auth = FiefAuth(fief, scheme)

FiefAuth( client: Union[fief_client.client.Fief, fief_client.client.FiefAsync], scheme: fastapi.security.base.SecurityBase, *, get_userinfo_cache: Optional[Callable[..., Union[UserInfoCacheProtocol, Coroutine[NoneType, NoneType, UserInfoCacheProtocol], AsyncGenerator[UserInfoCacheProtocol, NoneType], Generator[UserInfoCacheProtocol, NoneType, NoneType]]]] = None) View Source

105    def __init__(
106        self,
107        client: FiefClientClass,
108        scheme: SecurityBase,
109        *,
110        get_userinfo_cache: Optional[DependencyCallable[UserInfoCacheProtocol]] = None,
111    ) -> None:
112        """
113        :param client: Instance of a Fief client.
114        Can be either `fief_client.Fief` or `fief_client.FiefAsync`.
115        :param scheme: FastAPI security scheme.
116        It'll be used to retrieve the access token in the request.
117        :param get_userinfo_cache: Optional dependency returning an instance of a class
118        following the `UserInfoCacheProtocol`.
119        It'll be used to cache user information on your server.
120        Otherwise, the Fief API will always be reached when requesting user information.
121        """
122        self.client = client
123        self.scheme = scheme
124        self.get_userinfo_cache = get_userinfo_cache

Parameters

client: Instance of a Fief client. Can be either fief_client.Fief or fief_client.FiefAsync.
scheme: FastAPI security scheme. It'll be used to retrieve the access token in the request.
get_userinfo_cache: Optional dependency returning an instance of a class following the UserInfoCacheProtocol. It'll be used to cache user information on your server. Otherwise, the Fief API will always be reached when requesting user information.

client

scheme

get_userinfo_cache

def authenticated( self, optional: bool = False, scope: Optional[List[str]] = None, acr: Optional[fief_client.client.FiefACR] = None, permissions: Optional[List[str]] = None): View Source

126    def authenticated(
127        self,
128        optional: bool = False,
129        scope: Optional[List[str]] = None,
130        acr: Optional[FiefACR] = None,
131        permissions: Optional[List[str]] = None,
132    ):
133        """
134        Return a FastAPI dependency to check if a request is authenticated.
135
136        If the request is authenticated, the dependency will return a `fief_client.FiefAccessTokenInfo`.
137
138        :param optional: If `False` and the request is not authenticated,
139        an unauthorized response will be raised.
140        :param scope: Optional list of scopes required.
141        If the access token lacks one of the required scope, a forbidden response will be raised.
142        :param acr: Optional minimum ACR level required.
143        If the access token doesn't meet the minimum level, a forbidden response will be raised.
144        Read more: https://docs.fief.dev/going-further/acr/
145        :param permissions: Optional list of permissions required.
146        If the access token lacks one of the required permission, a forbidden response will be raised.
147
148        **Example**
149
150        ```py
151        @app.get("/authenticated")
152        async def get_authenticated(
153            access_token_info: FiefAccessTokenInfo = Depends(auth.authenticated()),
154        ):
155            return access_token_info
156        ```
157        """
158        signature = self._get_authenticated_call_signature(self.scheme)
159
160        @with_signature(signature)
161        async def _authenticated(
162            request: Request, response: Response, token: Optional[TokenType]
163        ) -> Optional[FiefAccessTokenInfo]:
164            if token is None:
165                if optional:
166                    return None
167                return await self.get_unauthorized_response(request, response)
168
169            if isinstance(token, HTTPAuthorizationCredentials):
170                token = token.credentials
171
172            try:
173                result = self.client.validate_access_token(
174                    token,
175                    required_scope=scope,
176                    required_acr=acr,
177                    required_permissions=permissions,
178                )
179                if isawaitable(result):
180                    info = await result
181                else:
182                    info = result
183            except (FiefAccessTokenInvalid, FiefAccessTokenExpired):
184                if optional:
185                    return None
186                return await self.get_unauthorized_response(request, response)
187            except (
188                FiefAccessTokenMissingScope,
189                FiefAccessTokenACRTooLow,
190                FiefAccessTokenMissingPermission,
191            ):
192                return await self.get_forbidden_response(request, response)
193
194            return info
195
196        return _authenticated

Return a FastAPI dependency to check if a request is authenticated.

If the request is authenticated, the dependency will return a fief_client.FiefAccessTokenInfo.

Parameters

optional: If False and the request is not authenticated, an unauthorized response will be raised.
scope: Optional list of scopes required. If the access token lacks one of the required scope, a forbidden response will be raised.
acr: Optional minimum ACR level required. If the access token doesn't meet the minimum level, a forbidden response will be raised. Read more: https://docs.fief.dev/going-further/acr/
permissions: Optional list of permissions required. If the access token lacks one of the required permission, a forbidden response will be raised.

Example

@app.get("/authenticated")
async def get_authenticated(
    access_token_info: FiefAccessTokenInfo = Depends(auth.authenticated()),
):
    return access_token_info

def current_user( self, optional: bool = False, scope: Optional[List[str]] = None, acr: Optional[fief_client.client.FiefACR] = None, permissions: Optional[List[str]] = None, refresh: bool = False): View Source

198    def current_user(
199        self,
200        optional: bool = False,
201        scope: Optional[List[str]] = None,
202        acr: Optional[FiefACR] = None,
203        permissions: Optional[List[str]] = None,
204        refresh: bool = False,
205    ):
206        """
207        Return a FastAPI dependency to check if a user is authenticated.
208
209        If the request is authenticated, the dependency will return a `fief_client.FiefUserInfo`.
210
211        If provided, the cache mechanism will be used to retrieve this information without calling the Fief API.
212
213        :param optional: If `False` and the request is not authenticated,
214        an unauthorized response will be raised.
215        :param scope: Optional list of scopes required.
216        If the access token lacks one of the required scope, a forbidden response will be raised.
217        :param acr: Optional minimum ACR level required.
218        If the access token doesn't meet the minimum level, a forbidden response will be raised.
219        Read more: https://docs.fief.dev/going-further/acr/
220        :param permissions: Optional list of permissions required.
221        If the access token lacks one of the required permission, a forbidden response will be raised.
222        :param refresh: If `True`, the user information will be refreshed from the Fief API.
223        Otherwise, the cache will be used.
224
225        **Example**
226
227        ```py
228        @app.get("/current-user", name="current_user")
229        async def get_current_user(
230            user: FiefUserInfo = Depends(auth.current_user()),
231        ):
232            return {"email": user["email"]}
233        ```
234        """
235        signature = self._get_current_user_call_signature(
236            self.authenticated(optional, scope, acr, permissions)
237        )
238
239        @with_signature(signature)
240        async def _current_user(
241            access_token_info: Optional[FiefAccessTokenInfo], *args, **kwargs
242        ) -> Optional[FiefUserInfo]:
243            userinfo_cache: Optional[UserInfoCacheProtocol] = kwargs.get(
244                "userinfo_cache"
245            )
246
247            if access_token_info is None and optional:
248                return None
249            assert access_token_info is not None
250
251            userinfo = None
252            if userinfo_cache is not None:
253                userinfo = await userinfo_cache.get(access_token_info["id"])
254
255            if userinfo is None or refresh:
256                result = self.client.userinfo(access_token_info["access_token"])
257                if isawaitable(result):
258                    userinfo = cast(FiefUserInfo, await result)
259                else:
260                    userinfo = cast(FiefUserInfo, result)
261
262                if userinfo_cache is not None:
263                    await userinfo_cache.set(access_token_info["id"], userinfo)
264
265            return userinfo
266
267        return _current_user

Return a FastAPI dependency to check if a user is authenticated.

If the request is authenticated, the dependency will return a fief_client.FiefUserInfo.

If provided, the cache mechanism will be used to retrieve this information without calling the Fief API.

Parameters

optional: If False and the request is not authenticated, an unauthorized response will be raised.
scope: Optional list of scopes required. If the access token lacks one of the required scope, a forbidden response will be raised.
acr: Optional minimum ACR level required. If the access token doesn't meet the minimum level, a forbidden response will be raised. Read more: https://docs.fief.dev/going-further/acr/
permissions: Optional list of permissions required. If the access token lacks one of the required permission, a forbidden response will be raised.
refresh: If True, the user information will be refreshed from the Fief API. Otherwise, the cache will be used.

Example

@app.get("/current-user", name="current_user")
async def get_current_user(
    user: FiefUserInfo = Depends(auth.current_user()),
):
    return {"email": user["email"]}

async def get_unauthorized_response( self, request: starlette.requests.Request, response: starlette.responses.Response): View Source

269    async def get_unauthorized_response(self, request: Request, response: Response):
270        """
271        Raise an `fastapi.HTTPException` with the status code 401.
272
273        This method is called when using the `authenticated` or `current_user` dependency
274        but the request is not authenticated.
275
276        You can override this method to customize the behavior in this case.
277        """
278        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

Raise an fastapi.HTTPException with the status code 401.

This method is called when using the authenticated or current_user dependency but the request is not authenticated.

You can override this method to customize the behavior in this case.

async def get_forbidden_response( self, request: starlette.requests.Request, response: starlette.responses.Response): View Source

280    async def get_forbidden_response(self, request: Request, response: Response):
281        """
282        Raise an `fastapi.HTTPException` with the status code 403.
283
284        This method is called when using the `authenticated` or `current_user` dependency
285        but the access token doesn't match the list of scopes or permissions.
286
287        You can override this method to customize the behavior in this case.
288        """
289        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)

Raise an fastapi.HTTPException with the status code 403.

This method is called when using the authenticated or current_user dependency but the access token doesn't match the list of scopes or permissions.

You can override this method to customize the behavior in this case.

FiefClientClass = typing.Union[fief_client.client.Fief, fief_client.client.FiefAsync]

class UserInfoCacheProtocol(typing.Protocol): View Source

53class UserInfoCacheProtocol(Protocol):
54    """
55    Protocol that should follow a class to implement a cache mechanism for user information.
56
57    Read more: https://docs.fief.dev/integrate/python/fastapi/#caching-user-information
58    """
59
60    async def get(self, user_id: uuid.UUID) -> Optional[FiefUserInfo]:
61        """
62        Retrieve user information from cache, if available.
63
64        :param user_id: The ID of the user to retrieve information for.
65        """
66        ...  # pragma: no cover
67
68    async def set(self, user_id: uuid.UUID, userinfo: FiefUserInfo) -> None:
69        """
70        Store user information in cache.
71
72        :param user_id: The ID of the user to cache information for.
73        :param userinfo: The user information to cache.
74        """
75        ...  # pragma: no cover

Protocol that should follow a class to implement a cache mechanism for user information.

UserInfoCacheProtocol(*args, **kwargs) View Source

1431def _no_init_or_replace_init(self, *args, **kwargs):
1432    cls = type(self)
1433
1434    if cls._is_protocol:
1435        raise TypeError('Protocols cannot be instantiated')
1436
1437    # Already using a custom `__init__`. No need to calculate correct
1438    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
1439    if cls.__init__ is not _no_init_or_replace_init:
1440        return
1441
1442    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
1443    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
1444    # searches for a proper new `__init__` in the MRO. The new `__init__`
1445    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
1446    # instantiation of the protocol subclass will thus use the new
1447    # `__init__` and no longer call `_no_init_or_replace_init`.
1448    for base in cls.__mro__:
1449        init = base.__dict__.get('__init__', _no_init_or_replace_init)
1450        if init is not _no_init_or_replace_init:
1451            cls.__init__ = init
1452            break
1453    else:
1454        # should not happen
1455        cls.__init__ = object.__init__
1456
1457    cls.__init__(self, *args, **kwargs)

async def get(self, user_id: uuid.UUID) -> Optional[fief_client.client.FiefUserInfo]: View Source

60    async def get(self, user_id: uuid.UUID) -> Optional[FiefUserInfo]:
61        """
62        Retrieve user information from cache, if available.
63
64        :param user_id: The ID of the user to retrieve information for.
65        """
66        ...  # pragma: no cover

Retrieve user information from cache, if available.

Parameters

user_id: The ID of the user to retrieve information for.

async def set( self, user_id: uuid.UUID, userinfo: fief_client.client.FiefUserInfo) -> None: View Source

68    async def set(self, user_id: uuid.UUID, userinfo: FiefUserInfo) -> None:
69        """
70        Store user information in cache.
71
72        :param user_id: The ID of the user to cache information for.
73        :param userinfo: The user information to cache.
74        """
75        ...  # pragma: no cover

Store user information in cache.

Parameters

user_id: The ID of the user to cache information for.
userinfo: The user information to cache.