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(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:

import databases
import sqlalchemy
from fastapi_users.db import (
    SQLAlchemyBaseOAuthAccountTable,
    SQLAlchemyBaseUserTable,
    SQLAlchemyUserDatabase,
)
from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base

from .models import UserDB

DATABASE_URL = "sqlite:///./test.db"
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__


async def get_user_db():
    yield SQLAlchemyUserDatabase(UserDB, database, users, oauth_accounts)

When instantiating the database adapter, you should pass this table in argument::

import databases
import sqlalchemy
from fastapi_users.db import (
    SQLAlchemyBaseOAuthAccountTable,
    SQLAlchemyBaseUserTable,
    SQLAlchemyUserDatabase,
)
from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base

from .models import UserDB

DATABASE_URL = "sqlite:///./test.db"
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__


async def get_user_db():
    yield SQLAlchemyUserDatabase(UserDB, database, users, oauth_accounts)

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 import models
from fastapi_users.db import TortoiseBaseOAuthAccountModel, TortoiseBaseUserModel
from tortoise import fields
from tortoise.contrib.pydantic import PydanticModel


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


class UserCreate(models.BaseUserCreate):
    pass


class UserUpdate(models.BaseUserUpdate):
    pass


class UserModel(TortoiseBaseUserModel):
    pass


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


class OAuthAccount(TortoiseBaseOAuthAccountModel):
    user = fields.ForeignKeyField("models.UserModel", 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:

from fastapi_users.db import TortoiseUserDatabase

from .models import OAuthAccount, UserDB, UserModel

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


async def get_user_db():
    yield TortoiseUserDatabase(UserDB, UserModel, OAuthAccount)

Generate a router¶

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

app.include_router(
  fastapi_users.get_oauth_router(google_oauth_client, auth_backend, "SECRET"),
  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¶

Open

requirements.txt main.py app/app.py app/db.py app/models.py app/users.py

fastapi
fastapi-users[sqlalchemy,oauth]
uvicorn[standard]

import uvicorn

if __name__ == "__main__":
    uvicorn.run("app.app:app", host="0.0.0.0", port=5000, log_level="info")

from fastapi import Depends, FastAPI

from app.db import database
from app.models import UserDB
from app.users import (
    auth_backend,
    current_active_user,
    fastapi_users,
    google_oauth_client,
)

app = FastAPI()

app.include_router(
    fastapi_users.get_auth_router(auth_backend), prefix="/auth/jwt", tags=["auth"]
)
app.include_router(fastapi_users.get_register_router(), prefix="/auth", tags=["auth"])
app.include_router(
    fastapi_users.get_reset_password_router(),
    prefix="/auth",
    tags=["auth"],
)
app.include_router(
    fastapi_users.get_verify_router(),
    prefix="/auth",
    tags=["auth"],
)
app.include_router(fastapi_users.get_users_router(), prefix="/users", tags=["users"])
app.include_router(
    fastapi_users.get_oauth_router(google_oauth_client, auth_backend, "SECRET"),
    prefix="/auth/google",
    tags=["auth"],
)


@app.get("/authenticated-route")
async def authenticated_route(user: UserDB = Depends(current_active_user)):
    return {"message": f"Hello {user.email}!"}


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


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

import databases
import sqlalchemy
from fastapi_users.db import (
    SQLAlchemyBaseOAuthAccountTable,
    SQLAlchemyBaseUserTable,
    SQLAlchemyUserDatabase,
)
from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base

from app.models import UserDB

DATABASE_URL = "sqlite:///./test.db"
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__


async def get_user_db():
    yield SQLAlchemyUserDatabase(UserDB, database, users, oauth_accounts)

from fastapi_users import models


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


class UserCreate(models.BaseUserCreate):
    pass


class UserUpdate(models.BaseUserUpdate):
    pass


class UserDB(User, models.BaseUserDB):
    pass

import os
from typing import Optional

from fastapi import Depends, Request
from fastapi_users import BaseUserManager, FastAPIUsers
from fastapi_users.authentication import (
    AuthenticationBackend,
    BearerTransport,
    JWTStrategy,
)
from fastapi_users.db import SQLAlchemyUserDatabase
from httpx_oauth.clients.google import GoogleOAuth2

from app.db import get_user_db
from app.models import User, UserCreate, UserDB, UserUpdate

SECRET = "SECRET"


google_oauth_client = GoogleOAuth2(
    os.environ["GOOGLE_OAUTH_CLIENT_ID"],
    os.environ["GOOGLE_OAUTH_CLIENT_SECRET"],
)


class UserManager(BaseUserManager[UserCreate, UserDB]):
    user_db_model = UserDB
    reset_password_token_secret = SECRET
    verification_token_secret = SECRET

    async def on_after_register(self, user: UserDB, request: Optional[Request] = None):
        print(f"User {user.id} has registered.")

    async def on_after_forgot_password(
        self, user: UserDB, token: str, request: Optional[Request] = None
    ):
        print(f"User {user.id} has forgot their password. Reset token: {token}")

    async def on_after_request_verify(
        self, user: UserDB, token: str, request: Optional[Request] = None
    ):
        print(f"Verification requested for user {user.id}. Verification token: {token}")


async def get_user_manager(user_db: SQLAlchemyUserDatabase = Depends(get_user_db)):
    yield UserManager(user_db)


bearer_transport = BearerTransport(tokenUrl="auth/jwt/login")


def get_jwt_strategy() -> JWTStrategy:
    return JWTStrategy(secret=SECRET, lifetime_seconds=3600)


auth_backend = AuthenticationBackend(
    name="jwt",
    transport=bearer_transport,
    get_strategy=get_jwt_strategy,
)
fastapi_users = FastAPIUsers(
    get_user_manager,
    [auth_backend],
    User,
    UserCreate,
    UserUpdate,
    UserDB,
)

current_active_user = fastapi_users.current_user(active=True)

MongoDB¶

Open

requirements.txt main.py app/app.py app/db.py app/models.py app/users.py

fastapi
fastapi-users[mongodb,oauth]
uvicorn[standard]

import uvicorn

if __name__ == "__main__":
    uvicorn.run("app.app:app", host="0.0.0.0", port=5000, log_level="info")

from fastapi import Depends, FastAPI

from app.models import UserDB
from app.users import (
    auth_backend,
    current_active_user,
    fastapi_users,
    google_oauth_client,
)

app = FastAPI()

app.include_router(
    fastapi_users.get_auth_router(auth_backend), prefix="/auth/jwt", tags=["auth"]
)
app.include_router(fastapi_users.get_register_router(), prefix="/auth", tags=["auth"])
app.include_router(
    fastapi_users.get_reset_password_router(),
    prefix="/auth",
    tags=["auth"],
)
app.include_router(
    fastapi_users.get_verify_router(),
    prefix="/auth",
    tags=["auth"],
)
app.include_router(fastapi_users.get_users_router(), prefix="/users", tags=["users"])
app.include_router(
    fastapi_users.get_oauth_router(google_oauth_client, auth_backend, "SECRET"),
    prefix="/auth/google",
    tags=["auth"],
)


@app.get("/authenticated-route")
async def authenticated_route(user: UserDB = Depends(current_active_user)):
    return {"message": f"Hello {user.email}!"}

import os

import motor.motor_asyncio
from fastapi_users.db import MongoDBUserDatabase

from app.models import UserDB

DATABASE_URL = os.environ["DATABASE_URL"]
client = motor.motor_asyncio.AsyncIOMotorClient(
    DATABASE_URL, uuidRepresentation="standard"
)
db = client["database_name"]
collection = db["users"]


async def get_user_db():
    yield MongoDBUserDatabase(UserDB, collection)

from fastapi_users import models


class User(models.BaseUser):
    pass


class UserCreate(models.BaseUserCreate):
    pass


class UserUpdate(models.BaseUserUpdate):
    pass


class UserDB(User, models.BaseUserDB):
    pass

import os
from typing import Optional

from fastapi import Depends, Request
from fastapi_users import BaseUserManager, FastAPIUsers
from fastapi_users.authentication import (
    AuthenticationBackend,
    BearerTransport,
    JWTStrategy,
)
from fastapi_users.db import MongoDBUserDatabase
from httpx_oauth.clients.google import GoogleOAuth2

from app.db import get_user_db
from app.models import User, UserCreate, UserDB, UserUpdate

SECRET = "SECRET"


google_oauth_client = GoogleOAuth2(
    os.environ["GOOGLE_OAUTH_CLIENT_ID"],
    os.environ["GOOGLE_OAUTH_CLIENT_SECRET"],
)


class UserManager(BaseUserManager[UserCreate, UserDB]):
    user_db_model = UserDB
    reset_password_token_secret = SECRET
    verification_token_secret = SECRET

    async def on_after_register(self, user: UserDB, request: Optional[Request] = None):
        print(f"User {user.id} has registered.")

    async def on_after_forgot_password(
        self, user: UserDB, token: str, request: Optional[Request] = None
    ):
        print(f"User {user.id} has forgot their password. Reset token: {token}")

    async def on_after_request_verify(
        self, user: UserDB, token: str, request: Optional[Request] = None
    ):
        print(f"Verification requested for user {user.id}. Verification token: {token}")


async def get_user_manager(user_db: MongoDBUserDatabase = Depends(get_user_db)):
    yield UserManager(user_db)


bearer_transport = BearerTransport(tokenUrl="auth/jwt/login")


def get_jwt_strategy() -> JWTStrategy:
    return JWTStrategy(secret=SECRET, lifetime_seconds=3600)


auth_backend = AuthenticationBackend(
    name="jwt",
    transport=bearer_transport,
    get_strategy=get_jwt_strategy,
)
fastapi_users = FastAPIUsers(
    get_user_manager,
    [auth_backend],
    User,
    UserCreate,
    UserUpdate,
    UserDB,
)

current_active_user = fastapi_users.current_user(active=True)

Tortoise ORM¶

Open

requirements.txt main.py app/app.py app/db.py app/models.py app/users.py

fastapi
fastapi-users[tortoise-orm,oauth]
uvicorn[standard]

import uvicorn

if __name__ == "__main__":
    uvicorn.run("app.app:app", host="0.0.0.0", port=5000, log_level="info")

from fastapi import Depends, FastAPI
from tortoise.contrib.fastapi import register_tortoise

from app.db import DATABASE_URL
from app.models import UserDB
from app.users import (
    auth_backend,
    current_active_user,
    fastapi_users,
    google_oauth_client,
)

app = FastAPI()

app.include_router(
    fastapi_users.get_auth_router(auth_backend), prefix="/auth/jwt", tags=["auth"]
)
app.include_router(fastapi_users.get_register_router(), prefix="/auth", tags=["auth"])
app.include_router(
    fastapi_users.get_reset_password_router(),
    prefix="/auth",
    tags=["auth"],
)
app.include_router(
    fastapi_users.get_verify_router(),
    prefix="/auth",
    tags=["auth"],
)
app.include_router(fastapi_users.get_users_router(), prefix="/users", tags=["users"])
app.include_router(
    fastapi_users.get_oauth_router(google_oauth_client, auth_backend, "SECRET"),
    prefix="/auth/google",
    tags=["auth"],
)


@app.get("/authenticated-route")
async def authenticated_route(user: UserDB = Depends(current_active_user)):
    return {"message": f"Hello {user.email}!"}


register_tortoise(
    app,
    db_url=DATABASE_URL,
    modules={"models": ["app.models"]},
    generate_schemas=True,
)

from fastapi_users.db import TortoiseUserDatabase

from app.models import OAuthAccount, UserDB, UserModel

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


async def get_user_db():
    yield TortoiseUserDatabase(UserDB, UserModel, OAuthAccount)

from fastapi_users import models
from fastapi_users.db import TortoiseBaseOAuthAccountModel, TortoiseBaseUserModel
from tortoise import fields
from tortoise.contrib.pydantic import PydanticModel


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


class UserCreate(models.BaseUserCreate):
    pass


class UserUpdate(models.BaseUserUpdate):
    pass


class UserModel(TortoiseBaseUserModel):
    pass


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


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

import os
from typing import Optional

from fastapi import Depends, Request
from fastapi_users import BaseUserManager, FastAPIUsers
from fastapi_users.authentication import (
    AuthenticationBackend,
    BearerTransport,
    JWTStrategy,
)
from fastapi_users.db import TortoiseUserDatabase
from httpx_oauth.clients.google import GoogleOAuth2

from app.db import get_user_db
from app.models import User, UserCreate, UserDB, UserUpdate

SECRET = "SECRET"


google_oauth_client = GoogleOAuth2(
    os.environ["GOOGLE_OAUTH_CLIENT_ID"],
    os.environ["GOOGLE_OAUTH_CLIENT_SECRET"],
)


class UserManager(BaseUserManager[UserCreate, UserDB]):
    user_db_model = UserDB
    reset_password_token_secret = SECRET
    verification_token_secret = SECRET

    async def on_after_register(self, user: UserDB, request: Optional[Request] = None):
        print(f"User {user.id} has registered.")

    async def on_after_forgot_password(
        self, user: UserDB, token: str, request: Optional[Request] = None
    ):
        print(f"User {user.id} has forgot their password. Reset token: {token}")

    async def on_after_request_verify(
        self, user: UserDB, token: str, request: Optional[Request] = None
    ):
        print(f"Verification requested for user {user.id}. Verification token: {token}")


async def get_user_manager(user_db: TortoiseUserDatabase = Depends(get_user_db)):
    yield UserManager(user_db)


bearer_transport = BearerTransport(tokenUrl="auth/jwt/login")


def get_jwt_strategy() -> JWTStrategy:
    return JWTStrategy(secret=SECRET, lifetime_seconds=3600)


auth_backend = AuthenticationBackend(
    name="jwt",
    transport=bearer_transport,
    get_strategy=get_jwt_strategy,
)
fastapi_users = FastAPIUsers(
    get_user_manager,
    [auth_backend],
    User,
    UserCreate,
    UserUpdate,
    UserDB,
)

current_active_user = fastapi_users.current_user(active=True)