import datetime
import jwt
from flask import current_app
class TokenError(Exception):
def __init__(self, reason):
super().__init__(reason)
self.reason = reason
def encode_auth_token(user_id, expires_in=3600):
now = datetime.datetime.utcnow()
payload = {
"sub": str(user_id),
"iat": now,
"exp": now + datetime.timedelta(seconds=expires_in),
}
return jwt.encode(payload, current_app.config["SECRET_KEY"], algorithm="HS256")
def decode_auth_token(token):
try:
payload = jwt.decode(
token, current_app.config["SECRET_KEY"], algorithms=["HS256"]
)
except jwt.ExpiredSignatureError:
raise TokenError("Token has expired")
except jwt.InvalidTokenError:
raise TokenError("Invalid authentication token")
subject = payload.get("sub")
if subject is None:
raise TokenError("Token is missing a subject claim")
return int(subject)
from functools import wraps
from flask import request, jsonify, g
from .models import User
from .tokens import decode_auth_token, TokenError
def _extract_bearer_token(header):
if not header:
return None
parts = header.split()
if len(parts) != 2 or parts[0].lower() != "bearer":
return None
return parts[1]
def token_required(view):
@wraps(view)
def wrapper(*args, **kwargs):
token = _extract_bearer_token(request.headers.get("Authorization"))
if token is None:
return jsonify(error="Missing or malformed Authorization header"), 401
try:
user_id = decode_auth_token(token)
except TokenError as exc:
return jsonify(error=exc.reason), 401
user = User.query.get(user_id)
if user is None or not user.is_active:
return jsonify(error="Account not found or disabled"), 401
g.current_user = user
return view(*args, **kwargs)
return wrapper
from flask import Blueprint, jsonify, g
from .auth import token_required
from .extensions import db
api = Blueprint("api", __name__, url_prefix="/api")
@api.get("/me")
@token_required
def profile():
user = g.current_user
return jsonify(id=user.id, email=user.email, name=user.name)
@api.post("/me/deactivate")
@token_required
def deactivate():
user = g.current_user
user.is_active = False
db.session.commit()
return jsonify(status="deactivated", id=user.id), 200
Token authentication in Flask usually boils down to two responsibilities that should stay separate: verifying a bearer token and making the resolved user available to the view without threading it through every argument. This snippet keeps those concerns apart by putting token encode/decode logic in one module and exposing the current user through Flask's request-local g object, which lives for exactly one request and is torn down afterwards.
In tokens.py, encode_auth_token builds a short-lived JWT with exp, iat, and a sub claim holding the user id. decode_auth_token is the mirror image and, crucially, it converts the two failure modes that PyJWT raises — ExpiredSignatureError and the broader InvalidTokenError — into a single TokenError with a human-readable reason. Collapsing library-specific exceptions into one domain exception means the decorator does not need to know which JWT library is in use, and the API returns consistent messages.
auth.py holds the token_required decorator. It first pulls the Authorization header and enforces the Bearer <token> shape with _extract_bearer_token; a malformed or missing header short-circuits with a 401 before any database work happens. The token is decoded, the sub claim is used to load the user, and a missing or inactive account is also rejected. The important line is g.current_user = user: because g is bound to the application context of the current request, the loaded user is now reachable from any code running later in that request without re-parsing the token or re-querying.
The decorator uses functools.wraps so the wrapped view keeps its name and docstring, which matters for Flask's URL-rule registration and for debugging. Errors are returned as JSON with explicit status codes rather than raised, keeping the API contract predictable.
routes.py shows the payoff. profile and deactivate simply declare @token_required and then read g.current_user. The views contain no auth logic at all, which is the whole point: authentication is enforced uniformly and the user is injected implicitly. A common pitfall this avoids is caching the user on a global — g is per-request, so there is no cross-request leakage even under a threaded or gevent server. The trade-off is that stateless JWTs cannot be revoked before expiry without an extra denylist check, so the tokens are kept short-lived.
Related snips
payload = {
sub: user.id,
iss: 'https://auth.example.com',
aud: 'codesnips-api',
exp: 15.minutes.from_now.to_i,
iat: Time.now.to_i,JWT issuance and verification without common footguns
#!/usr/bin/env bash
set -euo pipefail
export VAULT_ADDR="https://vault.internal:8200"
export VAULT_TOKEN="${VAULT_TOKEN:?missing VAULT_TOKEN}"Secrets management with environment isolation and Vault
import { randomBytes, createHash } from "crypto";
import jwt from "jsonwebtoken";
import { RefreshTokenStore } from "./store";
const ACCESS_SECRET = process.env.ACCESS_SECRET!;
const ACCESS_TTL = "15m";JWT access + refresh token rotation (conceptual)
package files
import (
"context"
"time"Presigned S3 upload URLs (AWS SDK v2)
<%# private stream: turbo signs the serialized record name %>
<%= turbo_stream_from current_user %>
<section class="notifications">
<h1>Notifications</h1>Turbo Streams + authorization: signed per-user stream name
package api
import (
"io"
"net/http"
"os"Safe multipart uploads using temp files (bounded memory)
Share this code
Here's the card — post it anywhere.
