Skip to content

OAuth2

FastAPI Users provides an optional OAuth2 authentication support. It relies on HTTPX OAuth library, which is a pure-async implementation of OAuth2.

Installation

You should install the library with the optional dependencies for OAuth:

pip install 'fastapi-users[sqlalchemy,oauth]'

pip install 'fastapi-users[mongodb,oauth]'

pip install 'fastapi-users[tortoise-orm,oauth]'

Configuration

Instantiate an OAuth2 client

You first need to get an HTTPX OAuth client instance. Read the documentation for more information.

from httpx_oauth.clients.google import GoogleOAuth2

google_oauth_client = GoogleOAuth2("CLIENT_ID", "CLIENT_SECRET")

Setup the models

The user models differ a bit from the standard one as we have to have a way to store the OAuth information (access tokens, account ids...).

from fastapi_users import models


class User(models.BaseUser, models.BaseOAuthAccountMixin):
    pass


class UserCreate(models.BaseUserCreate):
    pass


class UserUpdate(User, models.BaseUserUpdate):
    pass


class UserDB(User, models.BaseUserDB):
    pass

Notice that we inherit from the BaseOAuthAccountMixin, which adds a List of BaseOAuthAccount objects. This object is structured like this:

  • id (UUID4) – Unique identifier of the OAuth account information. Defaults to a UUID4.
  • oauth_name (str) – Name of the OAuth service. It corresponds to the name property of the OAuth client.
  • access_token (str) – Access token.
  • expires_at (Optional[int]) - Timestamp at which the access token is expired.
  • refresh_token (Optional[str]) – On services that support it, a token to get a fresh access token.
  • account_id (str) - Identifier of the OAuth account on the corresponding service.
  • account_email (str) - Email address of the OAuth account on the corresponding service.

Setup the database adapter

SQLAlchemy

You'll need to define the table for storing the OAuth account model. We provide a base one for this:

from fastapi_users.db.sqlalchemy import SQLAlchemyBaseOAuthAccountTable

class OAuthAccount(SQLAlchemyBaseOAuthAccountTable, Base):
    pass

Then, you should declare it on the database adapter:

user_db = SQLAlchemyUserDatabase(UserDB, database, User.__table__, OAuthAccount.__table__)

MongoDB

Nothing to do, the basic configuration is enough.

Tortoise ORM

You'll need to define the Tortoise model for storing the OAuth account model. We provide a base one for this:

from fastapi_users.db.tortoise import TortoiseBaseOAuthAccountModel


class OAuthAccount(TortoiseBaseOAuthAccountModel):
    user = fields.ForeignKeyField("models.User", related_name="oauth_accounts")

Warning

Note that you should define the foreign key yourself, so that you can point it the user model in your namespace.

Then, you should declare it on the database adapter:

user_db = TortoiseUserDatabase(UserDB, User, OAuthAccount)

Generate a router

Once you have a FastAPIUsers instance, you can make it generate a single OAuth router for the given client.

from fastapi import FastAPI
from fastapi_users import FastAPIUsers
from httpx_oauth.clients.google import GoogleOAuth2

google_oauth_client = GoogleOAuth2("CLIENT_ID", "CLIENT_SECRET")

app = FastAPI()
fastapi_users = FastAPIUsers(
    user_db, auth_backends, User, UserCreate, UserUpdate, UserDB
)

google_oauth_router = fastapi_users.get_oauth_router(google_oauth_client, SECRET)

app.include_router(google_oauth_router, prefix="/auth/google", tags=["auth"])

After register

You can provide a custom function to be called after a successful registration. It is called with two argument: the user that has just registered, and the original Request object.

Typically, you'll want to send a welcome e-mail or add it to your marketing analytics pipeline.

You can define it as an async or standard method.

Example:

from fastapi import FastAPI
from fastapi_users import FastAPIUsers
from httpx_oauth.clients.google import GoogleOAuth2


def on_after_register(user: UserDB, request: Request):
    print(f"User {user.id} has registered.")

google_oauth_client = GoogleOAuth2("CLIENT_ID", "CLIENT_SECRET")

app = FastAPI()
fastapi_users = FastAPIUsers(
    user_db, auth_backends, User, UserCreate, UserUpdate, UserDB
)

google_oauth_router = fastapi_users.get_oauth_router(google_oauth_client, SECRET, after_register=on_after_register)

app.include_router(google_oauth_router, prefix="/auth/google", tags=["auth"])

Full example

Warning

Notice that SECRET should be changed to a strong passphrase. Insecure passwords may give attackers full access to your database.

SQLAlchemy

import databases
import sqlalchemy
from fastapi import FastAPI, Request
from fastapi_users import FastAPIUsers, models
from fastapi_users.authentication import JWTAuthentication
from fastapi_users.db import (
    SQLAlchemyBaseOAuthAccountTable,
    SQLAlchemyBaseUserTable,
    SQLAlchemyUserDatabase,
)
from httpx_oauth.clients.google import GoogleOAuth2
from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base

DATABASE_URL = "sqlite:///./test.db"
SECRET = "SECRET"


google_oauth_client = GoogleOAuth2("CLIENT_ID", "CLIENT_SECRET")


class User(models.BaseUser, models.BaseOAuthAccountMixin):
    pass


class UserCreate(models.BaseUserCreate):
    pass


class UserUpdate(User, models.BaseUserUpdate):
    pass


class UserDB(User, models.BaseUserDB):
    pass


database = databases.Database(DATABASE_URL)
Base: DeclarativeMeta = declarative_base()


class UserTable(Base, SQLAlchemyBaseUserTable):
    pass


class OAuthAccount(SQLAlchemyBaseOAuthAccountTable, Base):
    pass


engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)
Base.metadata.create_all(engine)

users = UserTable.__table__
oauth_accounts = OAuthAccount.__table__
user_db = SQLAlchemyUserDatabase(UserDB, database, users, oauth_accounts)


def on_after_register(user: UserDB, request: Request):
    print(f"User {user.id} has registered.")


def on_after_forgot_password(user: UserDB, token: str, request: Request):
    print(f"User {user.id} has forgot their password. Reset token: {token}")


def after_verification_request(user: UserDB, token: str, request: Request):
    print(f"Verification requested for user {user.id}. Verification token: {token}")


jwt_authentication = JWTAuthentication(
    secret=SECRET, lifetime_seconds=3600, tokenUrl="auth/jwt/login"
)

app = FastAPI()
fastapi_users = FastAPIUsers(
    user_db,
    [jwt_authentication],
    User,
    UserCreate,
    UserUpdate,
    UserDB,
)
app.include_router(
    fastapi_users.get_auth_router(jwt_authentication), prefix="/auth/jwt", tags=["auth"]
)
app.include_router(
    fastapi_users.get_register_router(on_after_register), prefix="/auth", tags=["auth"]
)
app.include_router(
    fastapi_users.get_reset_password_router(
        SECRET, after_forgot_password=on_after_forgot_password
    ),
    prefix="/auth",
    tags=["auth"],
)
app.include_router(
    fastapi_users.get_verify_router(
        SECRET, after_verification_request=after_verification_request
    ),
    prefix="/auth",
    tags=["auth"],
)
app.include_router(fastapi_users.get_users_router(), prefix="/users", tags=["users"])

google_oauth_router = fastapi_users.get_oauth_router(
    google_oauth_client, SECRET, after_register=on_after_register
)
app.include_router(google_oauth_router, prefix="/auth/google", tags=["auth"])


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

MongoDB

import motor.motor_asyncio
from fastapi import FastAPI, Request
from fastapi_users import FastAPIUsers, models
from fastapi_users.authentication import JWTAuthentication
from fastapi_users.db import MongoDBUserDatabase
from httpx_oauth.clients.google import GoogleOAuth2

DATABASE_URL = "mongodb://localhost:27017"
SECRET = "SECRET"


google_oauth_client = GoogleOAuth2("CLIENT_ID", "CLIENT_SECRET")


class User(models.BaseUser, models.BaseOAuthAccountMixin):
    pass


class UserCreate(models.BaseUserCreate):
    pass


class UserUpdate(User, models.BaseUserUpdate):
    pass


class UserDB(User, models.BaseUserDB):
    pass


client = motor.motor_asyncio.AsyncIOMotorClient(
    DATABASE_URL, uuidRepresentation="standard"
)
db = client["database_name"]
collection = db["users"]
user_db = MongoDBUserDatabase(UserDB, collection)


def on_after_register(user: UserDB, request: Request):
    print(f"User {user.id} has registered.")


def on_after_forgot_password(user: UserDB, token: str, request: Request):
    print(f"User {user.id} has forgot their password. Reset token: {token}")


def after_verification_request(user: UserDB, token: str, request: Request):
    print(f"Verification requested for user {user.id}. Verification token: {token}")


jwt_authentication = JWTAuthentication(
    secret=SECRET, lifetime_seconds=3600, tokenUrl="auth/jwt/login"
)

app = FastAPI()
fastapi_users = FastAPIUsers(
    user_db,
    [jwt_authentication],
    User,
    UserCreate,
    UserUpdate,
    UserDB,
)
app.include_router(
    fastapi_users.get_auth_router(jwt_authentication), prefix="/auth/jwt", tags=["auth"]
)
app.include_router(
    fastapi_users.get_register_router(on_after_register), prefix="/auth", tags=["auth"]
)
app.include_router(
    fastapi_users.get_reset_password_router(
        SECRET, after_forgot_password=on_after_forgot_password
    ),
    prefix="/auth",
    tags=["auth"],
)
app.include_router(
    fastapi_users.get_verify_router(
        SECRET, after_verification_request=after_verification_request
    ),
    prefix="/auth",
    tags=["auth"],
)
app.include_router(fastapi_users.get_users_router(), prefix="/users", tags=["users"])

google_oauth_router = fastapi_users.get_oauth_router(
    google_oauth_client, SECRET, after_register=on_after_register
)
app.include_router(google_oauth_router, prefix="/auth/google", tags=["auth"])

Tortoise ORM

from fastapi import FastAPI, Request
from fastapi_users import FastAPIUsers, models
from fastapi_users.authentication import JWTAuthentication
from fastapi_users.db import (
    TortoiseBaseOAuthAccountModel,
    TortoiseBaseUserModel,
    TortoiseUserDatabase,
)
from httpx_oauth.clients.google import GoogleOAuth2
from tortoise import fields
from tortoise.contrib.fastapi import register_tortoise
from tortoise.contrib.pydantic import PydanticModel

DATABASE_URL = "sqlite://./test.db"
SECRET = "SECRET"


google_oauth_client = GoogleOAuth2("CLIENT_ID", "CLIENT_SECRET")


class UserModel(TortoiseBaseUserModel):
    pass


class OAuthAccountModel(TortoiseBaseOAuthAccountModel):
    user = fields.ForeignKeyField("models.UserModel", related_name="oauth_accounts")


class User(models.BaseUser, models.BaseOAuthAccountMixin):
    pass


class UserCreate(models.BaseUserCreate):
    pass


class UserUpdate(User, models.BaseUserUpdate):
    pass


class UserDB(User, models.BaseUserDB, PydanticModel):
    class Config:
        orm_mode = True
        orig_model = UserModel


user_db = TortoiseUserDatabase(UserDB, UserModel, OAuthAccountModel)
app = FastAPI()
register_tortoise(
    app,
    db_url=DATABASE_URL,
    modules={"models": ["oauth_full_tortoise"]},
    generate_schemas=True,
)


def on_after_register(user: UserDB, request: Request):
    print(f"User {user.id} has registered.")


def on_after_forgot_password(user: UserDB, token: str, request: Request):
    print(f"User {user.id} has forgot their password. Reset token: {token}")


def after_verification_request(user: UserDB, token: str, request: Request):
    print(f"Verification requested for user {user.id}. Verification token: {token}")


jwt_authentication = JWTAuthentication(
    secret=SECRET, lifetime_seconds=3600, tokenUrl="auth/jwt/login"
)

fastapi_users = FastAPIUsers(
    user_db,
    [jwt_authentication],
    User,
    UserCreate,
    UserUpdate,
    UserDB,
)
app.include_router(
    fastapi_users.get_auth_router(jwt_authentication), prefix="/auth/jwt", tags=["auth"]
)
app.include_router(
    fastapi_users.get_register_router(on_after_register), prefix="/auth", tags=["auth"]
)
app.include_router(
    fastapi_users.get_reset_password_router(
        SECRET, after_forgot_password=on_after_forgot_password
    ),
    prefix="/auth",
    tags=["auth"],
)
app.include_router(
    fastapi_users.get_verify_router(
        SECRET, after_verification_request=after_verification_request
    ),
    prefix="/auth",
    tags=["auth"],
)
app.include_router(fastapi_users.get_users_router(), prefix="/users", tags=["users"])

google_oauth_router = fastapi_users.get_oauth_router(
    google_oauth_client, SECRET, after_register=on_after_register
)
app.include_router(google_oauth_router, prefix="/auth/google", tags=["auth"])