Integrating FastAPI with Supabase Auth

Why this article and what we'll cover

In this article we'll go over the basics of how to implement login and logout functionality using Supabase Auth. We'll also go over how to guard routes so that only signed in users can access protected routes.

What is Supabase Auth

Supabase is a JSON Web Token based Auth service - it takes in the credentials of a user (for instance email and password) and returns a token that is used to securely transit information between parties. Other services can then make use of this token to know more about the user. For example, we can determine the user's role as well as the authentication methods that they have used to sign in.

Creating a sign up route

If integrating with Supabase Auth, one would simply need to wrap the client library in a route in order to get sign in, sign up, and sign out functionality. Here's an example:

Initialise a client in a centralised file

import os
from supabase import create_client, Client

url: str = os.environ.get("SUPABASE_URL")
key: str = os.environ.get("SUPABASE_KEY")
supa: Client = create_client(url, key)

Import the client into your application logic:

from .supabase import supa

@app.get("/sign_up")
def sign_up():
  res = supa.auth.sign_up(email="[e-mail]",
                          password="testsupabasenow")
  return res.get("access_token")

@app.get("/sign_out")
def sign_out():
  supa = init_supabase()
  res = supa.auth.sign_out()
  return "success"

@app.get("/sign_in")
def sign_in():
  supa = init_supabase()
  res = supa.auth.sign_in_with_password({"email":"[e-mail]", "password": "testsupabasenow"})
  return res.get("access_token")

To protect a route, there are two main options:

1. We can write a middleware function to validate the incoming JWT
2. We can use an extended version of the default FastAPI HTTPBearer together with Dependency Injection to validate the route

We're going to go with the second option as it is slightly less verbose.

Let's extend the built in HTTP Bearer class to decode and validate the JWT. We can do so by making use of a jwt library like python-jose or PyJWT. We'll also need the Supabase JWT secret which is under Settings > Auth at time of writing.

JWT Bearer class

Let's extend the built in JWT Bearer - below is a snippet taken from TestDriven.io's blog which performs a verification of the JWT to ensure that credentials are valid.

class JWTBearer(HTTPBearer):
    def __init__(self, auto_error: bool = True):
        super(JWTBearer, self).__init__(auto_error=auto_error)

    async def __call__(self, request: Request):
        credentials: HTTPAuthorizationCredentials = await super(JWTBearer, self).__call__(request)
        if credentials:
            if not credentials.scheme == "Bearer":
                raise HTTPException(status_code=403, detail="Invalid authentication scheme.")
            if not self.verify_jwt(credentials.credentials):
                raise HTTPException(status_code=403, detail="Invalid token or expired token.")
            return credentials.credentials
        else:
            raise HTTPException(status_code=403, detail="Invalid authorization code.")

    def verify_jwt(self, jwtoken: str) -> bool:
        isTokenValid: bool = False

        try:
            payload = decodeJWT(jwtoken)
        except:
            payload = None
        if payload:
            isTokenValid = True
        return isTokenValid

In order to protect the route, we can import the extended HTTP Bearer class and use the dependencies, option in FastAPI to protect the route.

Here's what it would look like:

from auth import JWTBearer
@app.post('/my-protected-route/',dependencies=[Depends(JWTBearer())], response_model=SchemaJob)
async def job(job: SchemaJob):
    ...
    # Some work here

Voila, you have a protected route.With this protection, requests to protected routes will raise an error if the Authorization Bearer: <jwt-token> header is not included.

It is also possible to extend this implementation by implementing custom checks based on claims or roles in def __call__ of the JWTBearer class above. For more information on JWTs, you may wish to check out the Supabase website Auth section for a deeper look into how JWTs function.

Here's the sample repo. Please feel free to leave any questions in the comments below.

The example code draws reference from:

1. https://testdriven.io/blog/fastapi-jwt-auth/
2. https://educative.io/answers/how-to-use-postgresql-database-in-fastapi

Comments

[Alex Leonov]

If you're using python-jose for jwt operations then use my code below. I found out that excluding the audience claim in jwt.decode() was causing errors.

from fastapi import HTTPException, Request
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose import JWTError, jwt

JWT_SECRET = "your-jwt-secret-here"

class JWTBearer(HTTPBearer):
    def __init__(self, auto_error: bool = True):
        super(JWTBearer, self).__init__(auto_error=auto_error)

    async def __call__(self, request: Request):
        credentials: HTTPAuthorizationCredentials = await super(JWTBearer, self).__call__(request)
        if credentials:
            if not credentials.scheme == "Bearer":
                raise HTTPException(status_code=401, detail="Invalid authentication scheme.")
            if not self.verify_jwt(credentials.credentials):
                raise HTTPException(status_code=401, detail="Invalid credentials.")
            return credentials.credentials
        else:
            raise HTTPException(status_code=401, detail="Invalid credentials.")

    def verify_jwt(self, jwtoken: str) -> bool:
        try:
            jwt.decode(jwtoken, JWT_SECRET, algorithms=["HS256"], audience="authenticated")
            return True
        except JWTError:
            return False

[Lee Yi Jie Joel]

Great catch! Thanks for sharing

[Peter Csiba]

I wish I found this article earlier thank you Lee!

Two notes:

1. Do you need to set the response cookie when you auth in FastAPI?

# response: fastapi.responses.Response
        response.set_cookie(
            key="access_token",
            value=f"Bearer {auth_response.session.access_token}",
            httponly=True,
        )

1. Really just a FYI that you can also connect directly to the GoTrue server / API (instead of the whole Supabase suite):

# pip install supabase-auth
from supabase_auth import SyncGoTrueClient, AuthResponse

    # https://github.com/supabase/auth-py/blob/main/README.md
    headers = {
        "apiKey": GOTRUE_JWT_SECRET,
    }
    client = SyncGoTrueClient(
        url=GOTRUE_URL,
        headers=headers,
    )
    auth_response = client.sign_up(email="[e-mail]", password="testsupabasenow")

Unsure of the benefits since supabase-py uses supabase-auth under the hood. For me personally it reduces the dependency on supabase and its possible blast radius.