fief_client

383class Fief(BaseFief):
384    """Sync Fief authentication client."""
385
386    def __init__(
387        self,
388        base_url: str,
389        client_id: str,
390        client_secret: Optional[str] = None,
391        *,
392        encryption_key: Optional[str] = None,
393        host: Optional[str] = None,
394        verify: VerifyTypes = True,
395        cert: Optional[CertTypes] = None,
396    ) -> None:
397        super().__init__(
398            base_url,
399            client_id,
400            client_secret,
401            encryption_key=encryption_key,
402            host=host,
403            verify=verify,
404            cert=cert,
405        )
406
407    def auth_url(
408        self,
409        redirect_uri: str,
410        *,
411        state: Optional[str] = None,
412        scope: Optional[List[str]] = None,
413        code_challenge: Optional[str] = None,
414        code_challenge_method: Optional[str] = None,
415        lang: Optional[str] = None,
416        extras_params: Optional[Mapping[str, str]] = None,
417    ) -> str:
418        """
419        Return an authorization URL.
420
421        :param redirect_uri: Your callback URI where the user will be redirected after Fief authentication.
422        :param state: Optional string that will be returned back in the callback parameters to allow you to retrieve state information.
423        :param scope: Optional list of scopes to ask for.
424        :param code_challenge: Optional code challenge for
425        [PKCE process](https://docs.fief.dev/going-further/pkce/).
426        :param code_challenge_method: Method used to hash the PKCE code challenge.
427        :param lang: Optional parameter to set the user locale.
428        Should be a valid [RFC 3066](https://www.rfc-editor.org/rfc/rfc3066) language identifier, like `fr` or `pt-PT`.
429        If not provided, the user locale is determined by their browser settings.
430        :param extras_params: Optional dictionary containing [specific parameters](https://docs.fief.dev/going-further/authorize-url/).
431
432        **Example:**
433
434        ```py
435        auth_url = fief.auth_url("http://localhost:8000/callback", scope=["openid"])
436        ```
437        """
438        openid_configuration = self._get_openid_configuration()
439        return self._auth_url(
440            openid_configuration,
441            redirect_uri,
442            state=state,
443            scope=scope,
444            code_challenge=code_challenge,
445            code_challenge_method=code_challenge_method,
446            lang=lang,
447            extras_params=extras_params,
448        )
449
450    def auth_callback(
451        self, code: str, redirect_uri: str, *, code_verifier: Optional[str] = None
452    ) -> Tuple[FiefTokenResponse, FiefUserInfo]:
453        """
454        Return a `FiefTokenResponse` and `FiefUserInfo` in exchange of an authorization code.
455
456        :param code: The authorization code.
457        :param redirect_uri: The exact same `redirect_uri` you passed to the authorization URL.
458        :param code_verifier:  The raw
459        [PKCE](https://docs.fief.dev/going-further/pkce/) code used to generate the code challenge during authorization.
460
461        **Example:**
462
463        ```py
464        tokens, userinfo = fief.auth_callback("CODE", "http://localhost:8000/callback")
465        ```
466        """
467        token_response = self._auth_exchange_token(
468            code, redirect_uri, code_verifier=code_verifier
469        )
470        jwks = self._get_jwks()
471        userinfo = self._decode_id_token(
472            token_response["id_token"],
473            jwks,
474            code=code,
475            access_token=token_response.get("access_token"),
476        )
477        return token_response, userinfo
478
479    def auth_refresh_token(
480        self, refresh_token: str, *, scope: Optional[List[str]] = None
481    ) -> Tuple[FiefTokenResponse, FiefUserInfo]:
482        """
483        Return fresh `FiefTokenResponse` and `FiefUserInfo` in exchange of a refresh token
484
485        :param refresh_token: A valid refresh token.
486        :param scope: Optional list of scopes to ask for.
487        If not provided, the access token will share the same list of scopes as requested the first time.
488        Otherwise, it should be a subset of the original list of scopes.
489
490        **Example:**
491
492        ```py
493        tokens, userinfo = fief.auth_refresh_token("REFRESH_TOKEN")
494        ```
495        """
496        token_endpoint = self._get_endpoint_url(
497            self._get_openid_configuration(), "token_endpoint"
498        )
499        with self._get_httpx_client() as client:
500            request = self._get_auth_refresh_token_request(
501                client,
502                endpoint=token_endpoint,
503                refresh_token=refresh_token,
504                scope=scope,
505            )
506            response = client.send(request)
507
508            self._handle_request_error(response)
509
510            token_response = response.json()
511        jwks = self._get_jwks()
512        userinfo = self._decode_id_token(
513            token_response["id_token"],
514            jwks,
515            access_token=token_response.get("access_token"),
516        )
517        return token_response, userinfo
518
519    def validate_access_token(
520        self,
521        access_token: str,
522        *,
523        required_scope: Optional[List[str]] = None,
524        required_permissions: Optional[List[str]] = None,
525    ) -> FiefAccessTokenInfo:
526        """
527        Check if an access token is valid and optionally that it has a required list of scopes,
528        or a required list of [permissions](https://docs.fief.dev/getting-started/access-control/).
529        Returns a `FiefAccessTokenInfo`.
530
531        :param access_token: The access token to validate.
532        :param required_scope: Optional list of scopes to check for.
533        :param required_permissions: Optional list of permissions to check for.
534
535        **Example: Validate access token with required scopes**
536
537        ```py
538        try:
539            access_token_info = fief.validate_access_token("ACCESS_TOKEN", required_scope=["required_scope"])
540        except FiefAccessTokenInvalid:
541            print("Invalid access token")
542        except FiefAccessTokenExpired:
543            print("Expired access token")
544        except FiefAccessTokenMissingScope:
545            print("Missing required scope")
546
547        print(access_token_info)
548        ```
549
550        **Example: Validate access token with required permissions**
551
552        ```py
553        try:
554            access_token_info = fief.validate_access_token("ACCESS_TOKEN", required_permissions=["castles:create", "castles:read"])
555        except FiefAccessTokenInvalid:
556            print("Invalid access token")
557        except FiefAccessTokenExpired:
558            print("Expired access token")
559        except FiefAccessTokenMissingPermission:
560            print("Missing required permission")
561
562        print(access_token_info)
563        ```
564        """
565        jwks = self._get_jwks()
566        return self._validate_access_token(
567            access_token,
568            jwks,
569            required_scope=required_scope,
570            required_permissions=required_permissions,
571        )
572
573    def userinfo(self, access_token: str) -> FiefUserInfo:
574        """
575        Return fresh `FiefUserInfo` from the Fief API using a valid access token.
576
577        :param access_token: A valid access token.
578
579        **Example:**
580
581        ```py
582        userinfo = fief.userinfo("ACCESS_TOKEN")
583        ```
584        """
585        userinfo_endpoint = self._get_endpoint_url(
586            self._get_openid_configuration(), "userinfo_endpoint"
587        )
588        with self._get_httpx_client() as client:
589            request = self._get_userinfo_request(
590                client, endpoint=userinfo_endpoint, access_token=access_token
591            )
592            response = client.send(request)
593
594            self._handle_request_error(response)
595
596            return response.json()
597
598    def update_profile(self, access_token: str, data: Dict[str, Any]) -> FiefUserInfo:
599        """
600        Update user information with the Fief API using a valid access token.
601
602        :param access_token: A valid access token.
603        :param data: A dictionary containing the data to update.
604
605        **Example: Update email address**
606
607        ```py
608        userinfo = fief.update_profile("ACCESS_TOKEN", { "email": "anne@nantes.city" })
609        ```
610
611        **Example: Update password**
612
613        ```py
614        userinfo = fief.update_profile("ACCESS_TOKEN", { "password": "herminetincture" })
615        ```
616
617        **Example: Update user field**
618
619        To update [user field](https://docs.fief.dev/getting-started/user-fields/) values, you need to nest them into a `fields` dictionary, indexed by their slug.
620
621        ```py
622        userinfo = fief.update_profile("ACCESS_TOKEN", { "fields": { "first_name": "Anne" } })
623        ```
624        """
625        update_profile_endpoint = f"{self.base_url}/api/profile"
626
627        with self._get_httpx_client() as client:
628            request = self._get_update_profile_request(
629                client,
630                endpoint=update_profile_endpoint,
631                access_token=access_token,
632                data=data,
633            )
634            response = client.send(request)
635
636            self._handle_request_error(response)
637
638            return response.json()
639
640    def logout_url(self, redirect_uri: str) -> str:
641        """
642        Returns a logout URL. If you redirect the user to this page, Fief will clear the session stored on its side.
643
644        **You're still responsible for clearing your own session mechanism if any.**
645
646        :param redirect_uri: A valid URL where the user will be redirected after the logout process.
647
648        **Example:**
649
650        ```py
651        logout_url = fief.logout_url("http://localhost:8000")
652        ```
653        """
654        params = {"redirect_uri": redirect_uri}
655        return f"{self.base_url}/logout?{urlencode(params)}"
656
657    @contextlib.contextmanager
658    def _get_httpx_client(self):
659        headers = {}
660        if self.host is not None:
661            headers["Host"] = self.host
662
663        with httpx.Client(
664            base_url=self.base_url, headers=headers, verify=self.verify, cert=self.cert
665        ) as client:
666            yield client
667
668    def _get_openid_configuration(self) -> Dict[str, Any]:
669        if self._openid_configuration is not None:
670            return self._openid_configuration
671
672        with self._get_httpx_client() as client:
673            request = self._get_openid_configuration_request(client)
674            response = client.send(request)
675            json = response.json()
676            self._openid_configuration = json
677            return json
678
679    def _get_jwks(self) -> jwk.JWKSet:
680        if self._jwks is not None:
681            return self._jwks
682
683        jwks_uri = self._get_endpoint_url(self._get_openid_configuration(), "jwks_uri")
684        with self._get_httpx_client() as client:
685            response = client.get(jwks_uri)
686            self._jwks = jwk.JWKSet.from_json(response.text)
687            return self._jwks
688
689    def _auth_exchange_token(
690        self, code: str, redirect_uri: str, *, code_verifier: Optional[str] = None
691    ) -> FiefTokenResponse:
692        token_endpoint = self._get_endpoint_url(
693            self._get_openid_configuration(), "token_endpoint"
694        )
695        with self._get_httpx_client() as client:
696            request = self._get_auth_exchange_token_request(
697                client,
698                endpoint=token_endpoint,
699                code=code,
700                redirect_uri=redirect_uri,
701                code_verifier=code_verifier,
702            )
703            response = client.send(request)
704
705            self._handle_request_error(response)
706
707            return response.json()

 710class FiefAsync(BaseFief):
 711    """Async Fief authentication client."""
 712
 713    def __init__(
 714        self,
 715        base_url: str,
 716        client_id: str,
 717        client_secret: Optional[str] = None,
 718        *,
 719        encryption_key: Optional[str] = None,
 720        host: Optional[str] = None,
 721        verify: VerifyTypes = True,
 722        cert: Optional[CertTypes] = None,
 723    ) -> None:
 724        super().__init__(
 725            base_url,
 726            client_id,
 727            client_secret,
 728            encryption_key=encryption_key,
 729            host=host,
 730            verify=verify,
 731            cert=cert,
 732        )
 733
 734    async def auth_url(
 735        self,
 736        redirect_uri: str,
 737        *,
 738        state: Optional[str] = None,
 739        scope: Optional[List[str]] = None,
 740        code_challenge: Optional[str] = None,
 741        code_challenge_method: Optional[str] = None,
 742        lang: Optional[str] = None,
 743        extras_params: Optional[Mapping[str, str]] = None,
 744    ) -> str:
 745        """
 746        Return an authorization URL.
 747
 748        :param redirect_uri: Your callback URI where the user will be redirected after Fief authentication.
 749        :param state: Optional string that will be returned back in the callback parameters to allow you to retrieve state information.
 750        :param scope: Optional list of scopes to ask for.
 751        :param code_challenge: Optional code challenge for
 752        [PKCE process](https://docs.fief.dev/going-further/pkce/).
 753        :param code_challenge_method: Method used to hash the PKCE code challenge.
 754        :param lang: Optional parameter to set the user locale.
 755        Should be a valid [RFC 3066](https://www.rfc-editor.org/rfc/rfc3066) language identifier, like `fr` or `pt-PT`.
 756        If not provided, the user locale is determined by their browser settings.
 757        :param extras_params: Optional dictionary containing [specific parameters](https://docs.fief.dev/going-further/authorize-url/).
 758
 759        **Example:**
 760
 761        ```py
 762        auth_url = await fief.auth_url("http://localhost:8000/callback", scope=["openid"])
 763        ```
 764        """
 765        openid_configuration = await self._get_openid_configuration()
 766        return self._auth_url(
 767            openid_configuration,
 768            redirect_uri,
 769            state=state,
 770            scope=scope,
 771            code_challenge=code_challenge,
 772            code_challenge_method=code_challenge_method,
 773            lang=lang,
 774            extras_params=extras_params,
 775        )
 776
 777    async def auth_callback(
 778        self, code: str, redirect_uri: str, *, code_verifier: Optional[str] = None
 779    ) -> Tuple[FiefTokenResponse, FiefUserInfo]:
 780        """
 781        Return a `FiefTokenResponse` and `FiefUserInfo` in exchange of an authorization code.
 782
 783        :param code: The authorization code.
 784        :param redirect_uri: The exact same `redirect_uri` you passed to the authorization URL.
 785        :param code_verifier:  The raw
 786        [PKCE](https://docs.fief.dev/going-further/pkce/) code used to generate the code challenge during authorization.
 787
 788        **Example:**
 789
 790        ```py
 791        tokens, userinfo = await fief.auth_callback("CODE", "http://localhost:8000/callback")
 792        ```
 793        """
 794        token_response = await self._auth_exchange_token(
 795            code, redirect_uri, code_verifier=code_verifier
 796        )
 797        jwks = await self._get_jwks()
 798        userinfo = self._decode_id_token(
 799            token_response["id_token"],
 800            jwks,
 801            code=code,
 802            access_token=token_response.get("access_token"),
 803        )
 804        return token_response, userinfo
 805
 806    async def auth_refresh_token(
 807        self, refresh_token: str, *, scope: Optional[List[str]] = None
 808    ) -> Tuple[FiefTokenResponse, FiefUserInfo]:
 809        """
 810        Return fresh `FiefTokenResponse` and `FiefUserInfo` in exchange of a refresh token
 811
 812        :param refresh_token: A valid refresh token.
 813        :param scope: Optional list of scopes to ask for.
 814        If not provided, the access token will share the same list of scopes as requested the first time.
 815        Otherwise, it should be a subset of the original list of scopes.
 816
 817        **Example:**
 818
 819        ```py
 820        tokens, userinfo = await fief.auth_refresh_token("REFRESH_TOKEN")
 821        ```
 822        """
 823        token_endpoint = self._get_endpoint_url(
 824            await self._get_openid_configuration(), "token_endpoint"
 825        )
 826        async with self._get_httpx_client() as client:
 827            request = self._get_auth_refresh_token_request(
 828                client,
 829                endpoint=token_endpoint,
 830                refresh_token=refresh_token,
 831                scope=scope,
 832            )
 833            response = await client.send(request)
 834
 835            self._handle_request_error(response)
 836
 837            token_response = response.json()
 838
 839        jwks = await self._get_jwks()
 840        userinfo = self._decode_id_token(
 841            token_response["id_token"],
 842            jwks,
 843            access_token=token_response.get("access_token"),
 844        )
 845        return token_response, userinfo
 846
 847    async def validate_access_token(
 848        self,
 849        access_token: str,
 850        *,
 851        required_scope: Optional[List[str]] = None,
 852        required_permissions: Optional[List[str]] = None,
 853    ) -> FiefAccessTokenInfo:
 854        """
 855        Check if an access token is valid and optionally that it has a required list of scopes,
 856        or a required list of [permissions](https://docs.fief.dev/getting-started/access-control/).
 857        Returns a `FiefAccessTokenInfo`.
 858
 859        :param access_token: The access token to validate.
 860        :param required_scope: Optional list of scopes to check for.
 861        :param required_permissions: Optional list of permissions to check for.
 862
 863        **Example: Validate access token with required scopes**
 864
 865        ```py
 866        try:
 867            access_token_info = await fief.validate_access_token("ACCESS_TOKEN", required_scope=["required_scope"])
 868        except FiefAccessTokenInvalid:
 869            print("Invalid access token")
 870        except FiefAccessTokenExpired:
 871            print("Expired access token")
 872        except FiefAccessTokenMissingScope:
 873            print("Missing required scope")
 874
 875        print(access_token_info)
 876        ```
 877
 878        **Example: Validate access token with required permissions**
 879
 880        ```py
 881        try:
 882            access_token_info = await fief.validate_access_token("ACCESS_TOKEN", required_permissions=["castles:create", "castles:read"])
 883        except FiefAccessTokenInvalid:
 884            print("Invalid access token")
 885        except FiefAccessTokenExpired:
 886            print("Expired access token")
 887        except FiefAccessTokenMissingPermission:
 888            print("Missing required permission")
 889
 890        print(access_token_info)
 891        ```
 892        """
 893        jwks = await self._get_jwks()
 894        return self._validate_access_token(
 895            access_token,
 896            jwks,
 897            required_scope=required_scope,
 898            required_permissions=required_permissions,
 899        )
 900
 901    async def userinfo(self, access_token: str) -> FiefUserInfo:
 902        """
 903        Return fresh `FiefUserInfo` from the Fief API using a valid access token.
 904
 905        :param access_token: A valid access token.
 906
 907        **Example:**
 908
 909        ```py
 910        userinfo = await fief.userinfo("ACCESS_TOKEN")
 911        ```
 912        """
 913        userinfo_endpoint = self._get_endpoint_url(
 914            await self._get_openid_configuration(), "userinfo_endpoint"
 915        )
 916        async with self._get_httpx_client() as client:
 917            request = self._get_userinfo_request(
 918                client, endpoint=userinfo_endpoint, access_token=access_token
 919            )
 920            response = await client.send(request)
 921
 922            self._handle_request_error(response)
 923
 924            return response.json()
 925
 926    async def update_profile(
 927        self, access_token: str, data: Dict[str, Any]
 928    ) -> FiefUserInfo:
 929        """
 930        Update user information with the Fief API using a valid access token.
 931
 932        :param access_token: A valid access token.
 933        :param data: A dictionary containing the data to update.
 934
 935        **Example: Update email address**
 936
 937        ```py
 938        userinfo = await fief.update_profile("ACCESS_TOKEN", { "email": "anne@nantes.city" })
 939        ```
 940
 941        **Example: Update password**
 942
 943        ```py
 944        userinfo = await fief.update_profile("ACCESS_TOKEN", { "password": "herminetincture" })
 945        ```
 946
 947        **Example: Update user field**
 948
 949        To update [user field](https://docs.fief.dev/getting-started/user-fields/) values, you need to nest them into a `fields` dictionary, indexed by their slug.
 950
 951        ```py
 952        userinfo = await fief.update_profile("ACCESS_TOKEN", { "fields": { "first_name": "Anne" } })
 953        ```
 954        """
 955        update_profile_endpoint = f"{self.base_url}/api/profile"
 956
 957        async with self._get_httpx_client() as client:
 958            request = self._get_update_profile_request(
 959                client,
 960                endpoint=update_profile_endpoint,
 961                access_token=access_token,
 962                data=data,
 963            )
 964            response = await client.send(request)
 965
 966            self._handle_request_error(response)
 967
 968            return response.json()
 969
 970    async def logout_url(self, redirect_uri: str) -> str:
 971        """
 972        Returns a logout URL. If you redirect the user to this page, Fief will clear the session stored on its side.
 973
 974        **You're still responsible for clearing your own session mechanism if any.**
 975
 976        :param redirect_uri: A valid URL where the user will be redirected after the logout process:
 977
 978        **Example:**
 979
 980        ```py
 981        logout_url = await fief.logout_url("http://localhost:8000")
 982        ```
 983        """
 984        params = {"redirect_uri": redirect_uri}
 985        return f"{self.base_url}/logout?{urlencode(params)}"
 986
 987    @contextlib.asynccontextmanager
 988    async def _get_httpx_client(self):
 989        headers = {}
 990        if self.host is not None:
 991            headers["Host"] = self.host
 992
 993        async with httpx.AsyncClient(
 994            base_url=self.base_url, headers=headers, verify=self.verify, cert=self.cert
 995        ) as client:
 996            yield client
 997
 998    async def _get_openid_configuration(self) -> Dict[str, Any]:
 999        if self._openid_configuration is not None:
1000            return self._openid_configuration
1001
1002        async with self._get_httpx_client() as client:
1003            request = self._get_openid_configuration_request(client)
1004            response = await client.send(request)
1005            json = response.json()
1006            self._openid_configuration = json
1007            return json
1008
1009    async def _get_jwks(self) -> jwk.JWKSet:
1010        if self._jwks is not None:
1011            return self._jwks
1012
1013        jwks_uri = self._get_endpoint_url(
1014            await self._get_openid_configuration(), "jwks_uri"
1015        )
1016        async with self._get_httpx_client() as client:
1017            response = await client.get(jwks_uri)
1018            self._jwks = jwk.JWKSet.from_json(response.text)
1019            return self._jwks
1020
1021    async def _auth_exchange_token(
1022        self, code: str, redirect_uri: str, *, code_verifier: Optional[str] = None
1023    ) -> FiefTokenResponse:
1024        token_endpoint = self._get_endpoint_url(
1025            await self._get_openid_configuration(), "token_endpoint"
1026        )
1027        async with self._get_httpx_client() as client:
1028            request = self._get_auth_exchange_token_request(
1029                client,
1030                endpoint=token_endpoint,
1031                code=code,
1032                redirect_uri=redirect_uri,
1033                code_verifier=code_verifier,
1034            )
1035            response = await client.send(request)
1036
1037            self._handle_request_error(response)
1038
1039            return response.json()