pip install hellojohn fastapi

import os
from hellojohn import HelloJohn

hj = HelloJohn(
    tenant_id=os.environ["HELLOJOHN_TENANT_ID"],
    secret_key=os.environ["HELLOJOHN_SECRET_KEY"],
)

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from hellojohn import InvalidTokenError, TokenPayload

security = HTTPBearer()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> TokenPayload:
    try:
        return hj.verify_token(credentials.credentials)
    except InvalidTokenError as e:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=str(e),
            headers={"WWW-Authenticate": "Bearer"},
        )

from fastapi import FastAPI

app = FastAPI()

@app.get("/api/me")
async def get_me(auth: TokenPayload = Depends(get_current_user)):
    return {"userId": auth.sub, "email": auth.email}

@app.post("/api/posts")
async def create_post(
    post: PostCreate,
    auth: TokenPayload = Depends(get_current_user),
    db: Session = Depends(get_db),
):
    return db.create_post(author_id=auth.sub, **post.model_dump())

from functools import partial

def require_role(role: str):
    async def check_role(auth: TokenPayload = Depends(get_current_user)):
        if auth.org_role not in (role, "owner"):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Insufficient permissions",
            )
        return auth
    return check_role

# Admin-only endpoint
@app.delete("/api/users/{user_id}")
async def delete_user(
    user_id: str,
    auth: TokenPayload = Depends(require_role("admin")),
):
    await hj.users.delete(user_id)
    return {"deleted": user_id}

from fastapi.security import HTTPBearer

optional_bearer = HTTPBearer(auto_error=False)

async def get_optional_user(
    credentials: HTTPAuthorizationCredentials | None = Depends(optional_bearer),
) -> TokenPayload | None:
    if not credentials:
        return None
    try:
        return hj.verify_token(credentials.credentials)
    except InvalidTokenError:
        return None

@app.get("/api/content")
async def get_content(auth: TokenPayload | None = Depends(get_optional_user)):
    if auth:
        return get_personalized_content(auth.sub)
    return get_public_content()

from fastapi import FastAPI, Depends, HTTPException, status
from contextlib import asynccontextmanager
from hellojohn import HelloJohn

hj = HelloJohn(
    tenant_id=os.environ["HELLOJOHN_TENANT_ID"],
    secret_key=os.environ["HELLOJOHN_SECRET_KEY"],
)

app = FastAPI(title="My API")

# Auth dependency
async def auth(credentials = Depends(HTTPBearer())):
    try:
        return hj.verify_token(credentials.credentials)
    except Exception:
        raise HTTPException(status_code=401, detail="Unauthorized")

# Routes
@app.get("/health")
def health():
    return {"ok": True}

@app.get("/api/me", dependencies=[Depends(auth)])
async def me(payload = Depends(auth)):
    user = await hj.users.get_async(payload.sub)
    return user