feat: add callback_raw to client

Author: everysoftware

README.md

@@ -25,7 +25,7 @@ pip install fastlink
 ## Get Started
 
 ```python
-from typing import Annotated
+from typing import Annotated, Any
 
 from fastapi import Depends, FastAPI
 from fastapi.responses import RedirectResponse
@@ -54,7 +54,6 @@ async def login() -> RedirectResponse:
 async def callback(call: Annotated[OAuth2Callback, Depends()]) -> OpenID:
     async with sso:
         return await sso.callback(call)
-
 ```
 
 Now you can run the server and visit `http://localhost:8000/login` to start the OAuth 2.0 flow.

examples/jwt.py

@@ -1,6 +1,7 @@
 from typing import Annotated
 
 from fastapi import Depends, FastAPI, Form, HTTPException, Response
+from starlette import status
 
 from fastlink.exceptions import FastLinkError
 from fastlink.integrations.fastapi.transport import CookieTransport
@@ -17,7 +18,7 @@
 @app.post("/login")
 def login(form: Annotated[OAuth2PasswordRequest, Form()]) -> Response:
     if form.username != "admin" or form.password != "admin":  # noqa: S105
-        raise HTTPException(status_code=400, detail="Incorrect username or password")
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Incorrect username or password")
     token = manager.create("access", JWTPayload(sub=form.username))
     return transport.set_token(Response(), token)
 
@@ -27,5 +28,5 @@ def get_me(token: Annotated[str, Depends(transport)]) -> JWTPayload:
     try:
         payload = manager.validate("access", token)
     except FastLinkError as e:
-        raise HTTPException(status_code=401, detail="Invalid token") from e
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token") from e
     return payload

fastlink/base/client.py

@@ -98,12 +98,12 @@ async def login_url(
 
     async def login(
         self,
-        callback: OAuth2Callback,
+        call: OAuth2Callback,
         *,
         body: dict[str, Any] | None = None,
         headers: dict[str, str] | None = None,
     ) -> TokenResponse:
-        request = self._prepare_token_request(callback, body=body, headers=headers)
+        request = self._prepare_token_request(call, body=body, headers=headers)
         auth = httpx.BasicAuth(self.client_id, self.client_secret)
         response = await self.client.send(
             request,
@@ -126,9 +126,13 @@ async def userinfo(self) -> dict[str, Any]:
             raise UserinfoError("Getting userinfo failed: %s", content)
         return cast("dict[str, Any]", content)
 
+    async def callback_raw(self, call: OAuth2Callback) -> dict[str, Any]:
+        await self.login(call)
+        return await self.userinfo()
+
     def _prepare_token_request(
         self,
-        callback: OAuth2Callback,
+        call: OAuth2Callback,
         *,
         body: dict[str, Any] | None = None,
         headers: dict[str, str] | None = None,
@@ -138,13 +142,13 @@ def _prepare_token_request(
         headers = headers or {}
         headers |= {"Content-Type": "application/x-www-form-urlencoded"}
         if self.use_state:
-            if not callback.state:
+            if not call.state:
                 raise StateError("State was not found in the callback")
-            body |= {"state": callback.state}
+            body |= {"state": call.state}
         body = {
             "grant_type": "authorization_code",
-            "code": callback.code,
-            "redirect_uri": callback.redirect_uri or self.redirect_uri,
+            "code": call.code,
+            "redirect_uri": call.redirect_uri or self.redirect_uri,
             "client_id": self.client_id,
             "client_secret": self.client_secret,
             **body,

fastlink/telegram/provider.py

@@ -101,6 +101,10 @@ async def userinfo(self) -> dict[str, Any]:
         assert self._callback is not None
         return self._callback.model_dump(exclude_none=True)
 
+    async def callback_raw(self, call: TelegramCallback) -> dict[str, Any]:
+        await self.login(call)
+        return await self.userinfo()
+
     async def openid_from_response(self, response: dict[str, Any]) -> OpenID:
         first_name, last_name = (
             response["first_name"],