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[sqlalchemy,oauth]'

    pip install 'fastapi-users[beanie,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 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, List

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

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

    class OAuthAccount(SQLAlchemyBaseOAuthAccountTableUUID, Base):
        pass

    class User(SQLAlchemyBaseUserTableUUID, Base):
        oauth_accounts: List[OAuthAccount] = relationship("OAuthAccount", lazy="joined")

    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(session, User, OAuthAccount)

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

Besides, when instantiating the database adapter, we need pass this SQLAlchemy model as third argument.

Primary key is defined as UUID

By default, we use UUID as a primary key ID for your user. If you want to use another type, like an auto-incremented integer, you can use `SQLAlchemyBaseOAuthAccountTable` as base class and define your own `id` and `user_id` column.

    class OAuthAccount(SQLAlchemyBaseOAuthAccountTable[int], Base):
        id = Column(Integer, primary_key=True)

        @declared_attr
        def user_id(cls):
            return Column(Integer, ForeignKey("user.id", ondelete="cascade"), nullable=False)

Notice that `SQLAlchemyBaseOAuthAccountTable` expects a generic type to define the actual type of ID you use.

#### Beanie

The advantage of MongoDB is that you can easily embed sub-objects in a single document. That's why the configuration for Beanie is quite simple. All we need to do is to define another class to structure an OAuth account object.

    from typing import List

    import motor.motor_asyncio
    from beanie import PydanticObjectId
    from fastapi_users.db import BaseOAuthAccount, BeanieBaseUser, BeanieUserDatabase
    from pydantic import Field

    DATABASE_URL = "mongodb://localhost:27017"
    client = motor.motor_asyncio.AsyncIOMotorClient(
        DATABASE_URL, uuidRepresentation="standard"
    )
    db = client["database_name"]

    class OAuthAccount(BaseOAuthAccount):
        pass

    class User(BeanieBaseUser[PydanticObjectId]):
        oauth_accounts: List[OAuthAccount] = Field(default_factory=list)

    async def get_user_db():
        yield BeanieUserDatabase(User)

It's worth to note that `OAuthAccount` is not a Beanie document but a Pydantic model that we'll embed inside the `User` document, through the `oauth_accounts` array.

### 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/schemas.pyapp/users.py

    fastapi
    fastapi-users[sqlalchemy]
    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 User, create_db_and_tables
    from app.schemas import UserCreate, UserRead, UserUpdate
    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(UserRead, UserCreate),
        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(UserRead),
        prefix="/auth",
        tags=["auth"],
    )
    app.include_router(
        fastapi_users.get_users_router(UserRead, UserUpdate),
        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: User = 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, List

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

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

    class OAuthAccount(SQLAlchemyBaseOAuthAccountTableUUID, Base):
        pass

    class User(SQLAlchemyBaseUserTableUUID, Base):
        oauth_accounts: List[OAuthAccount] = relationship("OAuthAccount", lazy="joined")

    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(session, User, OAuthAccount)

    import uuid

    from fastapi_users import schemas

    class UserRead(schemas.BaseUser[uuid.UUID]):
        pass

    class UserCreate(schemas.BaseUserCreate):
        pass

    class UserUpdate(schemas.BaseUserUpdate):
        pass

    import os
    import uuid
    from typing import Optional

    from fastapi import Depends, Request
    from fastapi_users import BaseUserManager, FastAPIUsers, UUIDIDMixin
    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 User, get_user_db

    SECRET = "SECRET"

    google_oauth_client = GoogleOAuth2(
        os.getenv("GOOGLE_OAUTH_CLIENT_ID", ""),
        os.getenv("GOOGLE_OAUTH_CLIENT_SECRET", ""),
    )

    class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
        reset_password_token_secret = SECRET
        verification_token_secret = SECRET

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

        async def on_after_forgot_password(
            self, user: User, 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: User, 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[User, uuid.UUID](get_user_manager, [auth_backend])

    current_active_user = fastapi_users.current_user(active=True)

#### Beanie

Open

requirements.txtmain.pyapp/app.pyapp/db.pyapp/schemas.pyapp/users.py

    fastapi
    fastapi-users[beanie]
    uvicorn[standard]

    import uvicorn

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

    from beanie import init_beanie
    from fastapi import Depends, FastAPI

    from app.db import User, db
    from app.schemas import UserCreate, UserRead, UserUpdate
    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(UserRead, UserCreate),
        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(UserRead),
        prefix="/auth",
        tags=["auth"],
    )
    app.include_router(
        fastapi_users.get_users_router(UserRead, UserUpdate),
        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: User = Depends(current_active_user)):
        return {"message": f"Hello {user.email}!"}

    @app.on_event("startup")
    async def on_startup():
        await init_beanie(
            database=db,
            document_models=[
                User,
            ],
        )

    from typing import List

    import motor.motor_asyncio
    from beanie import PydanticObjectId
    from fastapi_users.db import BaseOAuthAccount, BeanieBaseUser, BeanieUserDatabase
    from pydantic import Field

    DATABASE_URL = "mongodb://localhost:27017"
    client = motor.motor_asyncio.AsyncIOMotorClient(
        DATABASE_URL, uuidRepresentation="standard"
    )
    db = client["database_name"]

    class OAuthAccount(BaseOAuthAccount):
        pass

    class User(BeanieBaseUser[PydanticObjectId]):
        oauth_accounts: List[OAuthAccount] = Field(default_factory=list)

    async def get_user_db():
        yield BeanieUserDatabase(User)

    from beanie import PydanticObjectId
    from fastapi_users import schemas

    class UserRead(schemas.BaseUser[PydanticObjectId]):
        pass

    class UserCreate(schemas.BaseUserCreate):
        pass

    class UserUpdate(schemas.BaseUserUpdate):
        pass

    import os
    from typing import Optional

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

    from app.db import User, get_user_db

    SECRET = "SECRET"

    google_oauth_client = GoogleOAuth2(
        os.getenv("GOOGLE_OAUTH_CLIENT_ID", ""),
        os.getenv("GOOGLE_OAUTH_CLIENT_SECRET", ""),
    )

    class UserManager(ObjectIDIDMixin, BaseUserManager[User, PydanticObjectId]):
        reset_password_token_secret = SECRET
        verification_token_secret = SECRET

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

        async def on_after_forgot_password(
            self, user: User, 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: User, token: str, request: Optional[Request] = None
        ):
            print(f"Verification requested for user {user.id}. Verification token: {token}")

    async def get_user_manager(user_db: BeanieUserDatabase = 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[User, PydanticObjectId](get_user_manager, [auth_backend])

    current_active_user = fastapi_users.current_user(active=True)

Back to top