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

# Full example

Here is a full working example with JWT authentication to help get you started.

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]
    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

    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.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 SQLAlchemyBaseUserTable, SQLAlchemyUserDatabase
    from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
    from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
    from sqlalchemy.orm import sessionmaker

    from app.models import UserDB

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

    class UserTable(Base, SQLAlchemyBaseUserTable):
        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)

    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

    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 app.db import get_user_db
    from app.models import User, UserCreate, UserDB, UserUpdate

    SECRET = "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]
    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

    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.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

    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 app.db import get_user_db
    from app.models import User, UserCreate, UserDB, UserUpdate

    SECRET = "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]
    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

    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.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 UserDB, UserModel

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

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

    from fastapi_users import models
    from fastapi_users.db import TortoiseBaseUserModel
    from tortoise.contrib.pydantic import PydanticModel

    class User(models.BaseUser):
        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

    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 app.db import get_user_db
    from app.models import User, UserCreate, UserDB, UserUpdate

    SECRET = "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)

## What now?

You're ready to go! Be sure to check the Usage section to understand how to work with FastAPI Users.

Back to top