Skip to content

oauth

FractalOpenID

Bases: OpenID

Subclass of httpx_oauth.clients.openid.OpenID with customizable name for the "email" claim.

Source code in fractal_server/app/routes/auth/oauth.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class FractalOpenID(OpenID):
    """
    Subclass of `httpx_oauth.clients.openid.OpenID` with customizable name for
    the `"email"` claim.
    """

    def __init__(self, *, email_claim: str, **kwargs):
        super().__init__(**kwargs)
        self.email_claim = email_claim

    # TODO-requires-py312: add `@override` decorator
    async def get_id_email(self, token: str) -> tuple[str, str | None]:
        """
        Identical to the parent-class method (httpx-oauth version 0.16.1),
        apart from making `"email"` configurable.
        """
        try:
            profile = await self.get_profile(token)
        except GetProfileError as e:
            raise GetIdEmailError(response=e.response) from e
        return str(profile["sub"]), profile.get(self.email_claim)

get_id_email(token) async

Identical to the parent-class method (httpx-oauth version 0.16.1), apart from making "email" configurable.

Source code in fractal_server/app/routes/auth/oauth.py 
29
30
31
32
33
34
35
36
37
38
async def get_id_email(self, token: str) -> tuple[str, str | None]:
    """
    Identical to the parent-class method (httpx-oauth version 0.16.1),
    apart from making `"email"` configurable.
    """
    try:
        profile = await self.get_profile(token)
    except GetProfileError as e:
        raise GetIdEmailError(response=e.response) from e
    return str(profile["sub"]), profile.get(self.email_claim)

get_oauth_router()

Get the APIRouter object for OAuth endpoints.

Source code in fractal_server/app/routes/auth/oauth.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def get_oauth_router() -> APIRouter | None:
    """
    Get the `APIRouter` object for OAuth endpoints.
    """
    router_oauth = APIRouter()
    settings = Inject(get_settings)
    oauth_settings = Inject(get_oauth_settings)
    if not oauth_settings.is_set:
        return None

    client_name = oauth_settings.OAUTH_CLIENT_NAME
    if client_name == "google":
        client = _create_client_google(oauth_settings)
    elif client_name == "github":
        client = _create_client_github(oauth_settings)
    else:
        client = _create_client_oidc(oauth_settings)

    router_oauth.include_router(
        fastapi_users.get_oauth_router(
            client,
            cookie_backend,
            settings.JWT_SECRET_KEY,
            is_verified_by_default=False,
            associate_by_email=True,
            redirect_url=oauth_settings.OAUTH_REDIRECT_URL,
        ),
        prefix=f"/{client_name}",
    )

    # Add trailing slash to all routes' paths
    for route in router_oauth.routes:
        if not route.path.endswith("/"):
            route.path = f"{route.path}/"

    return router_oauth