FastAPI Login

---

1. Basic Usage with SQLAlchemy

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi_login import LoginManager

app = FastAPI()
login_manager = LoginManager(app, authenticate_user=authenticate_user, user_loader=load_user)

@login_manager.user_loader
def load_user(username: str) -> User:
    return get_user(username)

2. Customizing Authentication

from fastapi import FastAPI, Request
from fastapi_login import LoginManager, authenticate_user
from fastapi_login.exceptions import InvalidCredentialsException

app = FastAPI()
login_manager = LoginManager(app, authenticate_user=authenticate_user)

@login_manager.request_loader
def request_loader(request: Request) -> User:
    username = request.headers.get('Authorization')
    if username is None:
        raise InvalidCredentialsException
    return get_user(username)

3. Using Custom User Model

from fastapi import FastAPI
from fastapi_login import LoginManager, User

app = FastAPI()
login_manager = LoginManager(app)

@login_manager.user_loader
def load_user(username: str) -> User:
    user_data = get_user_data(username)
    return User(id=user_data.id, name=user_data.name, email=user_data.email)

4. Middleware to Inject Current User

from fastapi import FastAPI
from fastapi_login import LoginManager, user_loaded

app = FastAPI()
login_manager = LoginManager(app)

@app.middleware("http")
async def add_user_to_request(request: Request, call_next):
    try:
        user = await user_loaded(request)
    except Exception:
        user = None
    request.state.user = user
    return await call_next(request)

5. Role-Based Authorization

from fastapi import FastAPI
from fastapi_login import LoginManager, LoginRequired, current_user
from fastapi_login.exceptions import NotEnoughPermissionsException

app = FastAPI()
login_manager = LoginManager(app)

class User:
    def __init__(self, id: int, name: str, roles: list):
        self.id = id
        self.name = name
        self.roles = roles

@login_manager.user_loader
def load_user(username: str) -> User:
    return get_user(username)

@app.get("/protected", dependencies=[LoginRequired()])
async def protected_endpoint(user: User = current_user()):
    if "admin" not in user.roles:
        raise NotEnoughPermissionsException
    return {"message": "Hello, admin!"}

6. Refreshing JWT Tokens

from fastapi import FastAPI, Request, HTTPException
from fastapi_login import LoginManager, authenticate_user, login, JWTToken

app = FastAPI()
login_manager = LoginManager(app, authenticate_user=authenticate_user)

@login_manager.token_loader
def token_loader(request: Request) -> JWTToken | None:
    token = request.headers.get('Authorization')
    if token is None:
        return None
    return JWTToken(token)

@app.post("/refresh")
async def refresh_access_token(current_user: User = Depends(current_user())):
    access_token = await login(current_user)
    return {"access_token": access_token}

7. Using SQLAlchemy ORM with Multiple Tables

from fastapi import FastAPI
from fastapi_login import LoginManager, User, SQLAlchemy

app = FastAPI()
login_manager = LoginManager(app, authenticate_user=authenticate_user, user_loader=load_user)

user_model = SQLAlchemy(User, "users")
login_manager.init_app(app, user_model=user_model)

8. Customizing JWT Payload

from fastapi import FastAPI
from fastapi_login import JWTToken
from fastapi_login.utils import create_access_token

app = FastAPI()

@app.post("/login")
async def login(current_user: User = Depends(current_user())):
    access_token = create_access_token(
        subject=current_user.id,
        expires=timedelta(minutes=30),
        claims={"custom_claim": "custom_value"}
    )
    return {"access_token": access_token}

9. Using Redis for JWT Storage

from fastapi import FastAPI
from fastapi_login import JWTToken
from fastapi_login.storages import RedisStorage

app = FastAPI()

storage = RedisStorage(host="localhost", port=6379)
login_manager = LoginManager(app, token_storage=storage, authenticate_user=authenticate_user)

10. Using Cassandra for JWT Storage

from fastapi import FastAPI
from fastapi_login import JWTToken
from fastapi_login.storages import CassandraStorage

app = FastAPI()

storage = CassandraStorage(
    host="localhost",
    port=9042,
    keyspace_name="fastapi_login",
    table_name="jwt_tokens"
)
login_manager = LoginManager(app, token_storage=storage, authenticate_user=authenticate_user)

11. Using a Custom Storage Backend

from fastapi import FastAPI
from fastapi_login import JWTToken
from fastapi_login.storages.base import BaseStorage

class CustomStorage(BaseStorage):

    # Custom storage logic here

storage = CustomStorage()
login_manager = LoginManager(app, token_storage=storage, authenticate_user=authenticate_user)

12. Using a Background Task Queue

from fastapi import FastAPI
from fastapi_login import LoginManager
from fastapi_login.tasks import QueueTask

app = FastAPI()

task_queue = QueueTask()
login_manager = LoginManager(app, task_queue=task_queue, authenticate_user=authenticate_user)

13. Using a Dedicated Signing Key

from fastapi import FastAPI
from fastapi_login import LoginManager

app = FastAPI()
login_manager = LoginManager(app, signing_key="my-secret-key")

14. Using a Database for User Management

from fastapi import FastAPI
from fastapi_login import LoginManager, User
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class UserModel(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(255), unique=True)
    password = Column(String(255))

engine = create_engine("sqlite:///./users.db")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base.metadata.create_all(engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app = FastAPI()

login_manager = LoginManager(app, authenticate_user=authenticate_user, user_loader=load_user)

@login_manager.user_loader
def load_user(username: str) -> User | None:
    with get_db() as db:
        user = db.query(UserModel).filter(UserModel.username == username).first()
        return User(id=user.id, username=user.username) if user else None

15. Using a Password Hashing Algorithm

from fastapi import FastAPI
from fastapi_login import LoginManager, User
from passlib.context import CryptContext

app = FastAPI()

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
login_manager = LoginManager(app, authenticate_user=authenticate_user, user_loader=load_user, password_hasher=pwd_context)

@login_manager.user_loader
def load_user(username: str) -> User | None:
    user_data = get_user_data(username)
    return User(id=user_data.id, username=user_data.username, hashed_password=user_data.hashed_password) if user_data else None

16. Using a Password Reset Feature

from fastapi import FastAPI
from fastapi_login import LoginManager, User, PasswordReset
from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.orm import sessionmaker, aliased
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class UserModel(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(255), unique=True)
    hashed_password = Column(String(255))

class PasswordResetModel(Base):
    __tablename__ = "password_resets"
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("users.id"))
    token = Column(String(255))
    created_at = Column(DateTime)

engine = create_engine("sqlite:///./users.db")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base.metadata.create_all(engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app = FastAPI()
login_manager = LoginManager(app, authenticate_user=authenticate_user, user_loader=load_user)
password_reset = PasswordReset(app, token_storage=RedisStorage())

@login_manager.user_loader
def load_user(username: str) -> User | None:
    with get_db() as db:
        user = db.query(UserModel).filter(UserModel.username == username).first()
        return User(id=user.id, username=user.username, hashed_password=user.hashed_password) if user else None

17. Using a Custom Error Handler

from fastapi import FastAPI
from fastapi_login import LoginManager, LoginError, HTTPUnauthorizedException

app = FastAPI()
login_manager = LoginManager(app, authenticate_user=authenticate_user, error_handler=custom_error_handler)

def custom_error_handler(exception: LoginError) -> HTTPUnauthorizedException:
    return HTTPUnauthorizedException(headers={"WWW-Authenticate": "Bearer"})

18. Using a JSON Web Token (JWT) for Authentication

from fastapi import FastAPI
from fastapi_login import LoginManager
from fastapi_login.exceptions import InvalidCredentialsException
from fastapi_login.utils import create_access_token

app = FastAPI()
login_manager = LoginManager(app, authenticate_user=authenticate_user)

@login_manager.request_loader
def request_loader(request: Request) -> User:
    token = request.headers.get("Authorization")
    if token is None:
        raise InvalidCredentialsException
    user = load_user_from_token(token)
    return user

19. Using a Bearer Token Authentication Scheme

from fastapi import FastAPI, Depends
from fastapi.security import OAuth2PasswordBearer
from fastapi_login import LoginManager, current_user
from fastapi_login.exceptions import InvalidCredentialsException
from fastapi_login.utils import create_access_token

app = FastAPI()
login_manager = LoginManager(app, authenticate_user=authenticate_user)

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/login")

@app.post("/login")
async def login(username: str, password: str):
    user = await login_manager.authenticate_user(username, password)
    access_token = create_access_token(sub=user.id)
    return {"access_token": access_token}

@app.get("/protected", dependencies=[Depends(oauth2_scheme), Depends(current_user)])
async def protected_endpoint(user: User = current_user()):
    return {"message": f"Hello, {user.username}!"}

20. Using a Scoped Bearer Token Authentication Scheme

from fastapi import FastAPI, Depends
from fastapi.security import OAuth2PasswordBearer
from fastapi_login import LoginManager, current_user
from fastapi_login.exceptions import InvalidCredentialsException, TokenExpired
from fastapi_login.utils import create_access_token
from fastapi_login.security import Scopes

app = FastAPI()
login_manager = LoginManager(app, authenticate_user=authenticate_user)

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/login", scopes={"admin": "Admin Scope", "user": "User Scope"})

@app.post("/login")
async def login(username: str, password: str, scopes: Scopes = Depends(oauth2_scheme)):
    user = await login_manager.authenticate_user(username, password)
    access_token = create_access_token(sub=user.id, scopes=scopes.scope)
    return {"access_token": access_token}

@app.get("/protected-admin", dependencies=[Depends(oauth2_scheme), Depends(current_user), Depends(scopes.admin)])
async def protected_admin_endpoint(user: User = current_user()):
    return {"message": f"Hello, {user.username}! You have admin privileges."}

@app.get("/protected-user", dependencies=[Depends(oauth2_scheme), Depends(current_user), Depends(scopes.user)])
async def protected_user_endpoint(user: User = current_user()):
    return {"message": f"Hello, {user.username}! You have user privileges."}

21. Using a CSRF Protection Middleware

from fastapi import FastAPI
from fastapi.middleware.csrf import CSRFMiddleware

app = FastAPI()
app.add_middleware(
    CSRFMiddleware,
    csrf_token_header="X-CSRFToken",
)

22. Using a Throttling Middleware

from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddleware

app = FastAPI()
app.add_middleware(
    GZipMiddleware,
)

23. Using a Content Security Policy (CSP) Middleware

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

24. Using a CORS Middleware

from fastapi import FastAPI
from fastapi.middleware.security.basic import BasicAuth

app = FastAPI()
app.add_middleware(
    BasicAuth,
    username="user",
    password="password",
)

25. Using a Basic Authentication Middleware

from fastapi import FastAPI
from fastapi.middleware.sessions import SessionMiddleware

app = FastAPI()
app.add_middleware(
    SessionMiddleware,
    secret_key="my-secret-key",
)

26. Using a Session Middleware

from fastapi import FastAPI
from fastapi.middleware.trustedhost import TrustedHostMiddleware

app = FastAPI()
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["example.com", "www.example.com"],
)

27. Using a Trusted Host Middleware

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")

28. Using a Static Files Middleware

from fastapi import FastAPI
from fastapi.exceptions import HTTPException
from fastapi.responses import HTMLResponse

app = FastAPI()

@app.get("/custom-error-handler", status_code=404)
async def custom_404_handler():
    return HTMLResponse(content="<h1>Not Found</h1>", status_code=404)

29. Using a Custom Error Handler for 404

from fastapi import FastAPI
from fastapi.exceptions import HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(HTTPException)
async def generic_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse({"error": str(exc)})

30. Using a Generic Exception Handler

from fastapi import FastAPI
from fastapi.responses import Response

app = FastAPI()

@app.middleware("http")
async def add_custom_header(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-Custom-Header"] = "value"
    return response

31. Using a Custom Middleware to Add a Header

from fastapi import FastAPI
from fastapi.logger import logger

app = FastAPI()

@app.middleware("http")
async def log_request_payload(request: Request, call_next):
    try:
        request_body = await request.json()
        logger.info("Request payload: %s", request_body)
    except Exception:
        logger.info("Request payload could not be parsed as JSON")
    response = await call_next(request)
    return response

32. Using a Custom Middleware to Log Request Payload

from fastapi import FastAPI
from starlette.responses import RedirectResponse

app = FastAPI()

@app.get("/")
async def root():
    return RedirectResponse("/docs")

33. Using a Redirect Response

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/stream")
async def streaming():
    def generate_data():
        for i in range(10):
            yield f"Line {i}\n"
    return StreamingResponse(generate_data())

34. Using a Streaming Response

from fastapi import FastAPI, Response
from fastapi.responses import FileResponse

app = FastAPI()

@app.get("/file")
async def file():
    return FileResponse("path/to/file.txt")

35. Using a File Response

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse

app = FastAPI()

@app.get("/html")
async def html(request: Request):
    return HTMLResponse(content="<h1>Hello, world!</h1>", status_code=200)

36. Using an HTML Response