Skip to content
You're not viewing the latest version. Click here to go to latest.

# 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[sqlalchemy2,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 SQLAlchemy model for storing OAuth accounts. We provide a base one for this:

    from typing import AsyncGenerator

    from fastapi import Depends
    from fastapi_users.db import (
        SQLAlchemyBaseOAuthAccountTable,
        SQLAlchemyBaseUserTable,
        SQLAlchemyUserDatabase,
    )
    from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
    from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
    from sqlalchemy.orm import relationship, sessionmaker

    from .models import UserDB

    DATABASE_URL = "sqlite+aiosqlite:///./test.db"
    Base: DeclarativeMeta = declarative_base()

    class UserTable(Base, SQLAlchemyBaseUserTable):
        oauth_accounts = relationship("OAuthAccountTable")

    class OAuthAccountTable(SQLAlchemyBaseOAuthAccountTable, Base):
        pass

    engine = create_async_engine(DATABASE_URL)
    async_session_maker = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

    async def create_db_and_tables():
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)

    async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
        async with async_session_maker() as session:
            yield session

    async def get_user_db(session: AsyncSession = Depends(get_async_session)):
        yield SQLAlchemyUserDatabase(UserDB, session, UserTable, OAuthAccountTable)

Notice that we also manually added a `relationship` on the `UserTable` so that SQLAlchemy can properly retrieve the OAuth accounts of the user.

When instantiating the database adapter, you should pass this SQLAlchemy model:

    from typing import AsyncGenerator

    from fastapi import Depends
    from fastapi_users.db import (
        SQLAlchemyBaseOAuthAccountTable,
        SQLAlchemyBaseUserTable,
        SQLAlchemyUserDatabase,
    )
    from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
    from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
    from sqlalchemy.orm import relationship, sessionmaker

    from .models import UserDB

    DATABASE_URL = "sqlite+aiosqlite:///./test.db"
    Base: DeclarativeMeta = declarative_base()

    class UserTable(Base, SQLAlchemyBaseUserTable):
        oauth_accounts = relationship("OAuthAccountTable")

    class OAuthAccountTable(SQLAlchemyBaseOAuthAccountTable, Base):
        pass

    engine = create_async_engine(DATABASE_URL)
    async_session_maker = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

    async def create_db_and_tables():
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)

    async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
        async with async_session_maker() as session:
            yield session

    async def get_user_db(session: AsyncSession = Depends(get_async_session)):
        yield SQLAlchemyUserDatabase(UserDB, session, UserTable, OAuthAccountTable)

#### 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.txtmain.pyapp/app.pyapp/db.pyapp/models.pyapp/users.py

    fastapi
    fastapi-users[sqlalchemy2,oauth]
    uvicorn[standard]
    aiosqlite

    import uvicorn

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

    from fastapi import Depends, FastAPI

    from app.db import create_db_and_tables
    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 on_startup():
        # Not needed if you setup a migration system like Alembic
        await create_db_and_tables()

    from typing import AsyncGenerator

    from fastapi import Depends
    from fastapi_users.db import (
        SQLAlchemyBaseOAuthAccountTable,
        SQLAlchemyBaseUserTable,
        SQLAlchemyUserDatabase,
    )
    from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
    from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
    from sqlalchemy.orm import relationship, sessionmaker

    from app.models import UserDB

    DATABASE_URL = "sqlite+aiosqlite:///./test.db"
    Base: DeclarativeMeta = declarative_base()

    class UserTable(Base, SQLAlchemyBaseUserTable):
        oauth_accounts = relationship("OAuthAccountTable")

    class OAuthAccountTable(SQLAlchemyBaseOAuthAccountTable, Base):
        pass

    engine = create_async_engine(DATABASE_URL)
    async_session_maker = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

    async def create_db_and_tables():
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)

    async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
        async with async_session_maker() as session:
            yield session

    async def get_user_db(session: AsyncSession = Depends(get_async_session)):
        yield SQLAlchemyUserDatabase(UserDB, session, UserTable, OAuthAccountTable)

    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.getenv("GOOGLE_OAUTH_CLIENT_ID", ""),
        os.getenv("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.txtmain.pyapp/app.pyapp/db.pyapp/models.pyapp/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, 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 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.txtmain.pyapp/app.pyapp/db.pyapp/models.pyapp/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)

Back to top