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