Skip to content

Oauth

CLASS DESCRIPTION
FractalOpenID

Subclass of httpx_oauth.clients.openid.OpenID with customizable name for

FUNCTION DESCRIPTION
get_oauth_router

Get the APIRouter object for OAuth endpoints.

Attributes

Classes

FractalOpenID

Bases: OpenID

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

METHOD DESCRIPTION
get_id_email

Identical to the parent-class method (httpx-oauth version 0.16.1),

Source code in fractal_server/app/routes/auth/oauth.py
class FractalOpenID(OpenID):
    """
    Subclass of `httpx_oauth.clients.openid.OpenID` with customizable name for
    the `"email"` claim.
    """

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

    @override
    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)

Methods:

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
@override
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)

Functions:

get_oauth_router()

Get the APIRouter object for OAuth endpoints.

Source code in fractal_server/app/routes/auth/oauth.py
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_in_place(router_oauth)
    return router_oauth