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

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

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

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

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

    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 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 app.db import User, get_user_db

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

    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.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,
            ],
        )

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

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

    class User(BeanieBaseUser[PydanticObjectId]):
        pass

    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

    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 app.db import User, get_user_db

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

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