auth-pipeline (Python / FastAPI)

A unified, strategy-based auth system for FastAPI. Google OAuth, credentials, or anything else — every strategy resolves to the same result shape, and you decide what happens next in one place.

This is the Python port of the JS auth-kit package — same architecture, same unified shape, ported to FastAPI's async/dependency-injection style.

pip install -e .

---

The core idea

Authenticating with Google and authenticating with email/password are fundamentally different processes — but what your server does after either succeeds is usually identical: look up or create a user, issue a session, redirect or respond. auth-pipeline splits this into two layers:

Layer	Responsibility	Where it lives
Strategy	Talk to the provider, return a normalized user	auth_pipeline/strategies/*.py
AuthKit	Wire strategies to routes, call your hook	auth_pipeline/core/auth_kit.py
Your hook	Decide what "logged in" means for your app	your code

Every strategy resolves to exactly this (AuthResult):

AuthResult(
    provider="google",       # or "credentials", "github", ...
    user=AuthUser(id=..., email=..., name=..., picture=..., email_verified=...),
    raw={...},                # untouched provider payload, for anything custom
)

Your on_authenticate hook receives this every time, regardless of which strategy ran. Add a GitHub strategy next month — your hook doesn't change.

---

Quick start

import jwt
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, RedirectResponse, Response
from auth_pipeline import AuthKit, AuthResult, GoogleStrategy, CredentialsStrategy

app = FastAPI()

google = GoogleStrategy(
    client_id=os.environ["GOOGLE_CLIENT_ID"],
    client_secret=os.environ["GOOGLE_CLIENT_SECRET"],
    redirect_uri="http://localhost:8000/auth/google/callback",
)

async def verify_user(email: str, password: str, request: Request):
    row = await db.users.find_by_email(email)         # your DB, your choice
    if not row:
        return None
    if not CredentialsStrategy.verify_password(password, row["password_hash"]):
        return None
    return row

credentials = CredentialsStrategy(verify=verify_user)

async def on_authenticate(result: AuthResult, request: Request) -> Response:
    user = result.user
    token = jwt.encode({"id": user.id, "email": user.email}, JWT_SECRET, algorithm="HS256")

    if result.provider == "google":
        return RedirectResponse(f"http://localhost:3000/auth/callback?token={token}")
    return JSONResponse({"token": token, "user": user.to_dict()})

auth_kit = AuthKit(
    strategies={"google": google, "credentials": credentials},
    on_authenticate=on_authenticate,
)

@app.get("/auth/google")
async def google_start(request: Request):
    return await auth_kit.initiate("google", request)

@app.get("/auth/google/callback")
async def google_callback(request: Request):
    return await auth_kit.complete("google", request)

@app.post("/auth/login")
async def login(request: Request):
    return await auth_kit.complete("credentials", request)

Notice: initiate() and complete() are the same two calls for every strategy — only the string changes.

---

Strategies included

GoogleStrategy

GoogleStrategy(
    client_id,        # required
    client_secret,    # required
    redirect_uri,      # required
    scopes=("email", "profile"),
    access_type="offline",
    prompt="consent",
)

result.raw contains {"profile": ..., "tokens": ...} — the full Google profile and token response, in case you need fields beyond the normalized set (e.g. locale, or Google's refresh_token to call Google APIs later).

CredentialsStrategy

CredentialsStrategy(
    verify=my_verify_fn,   # async def(email: str, password: str, request: Request) -> dict | None
    email_field="email",
    password_field="password",
)

Like the JS version, this strategy does not touch your database. You provide verify; it handles request parsing, password comparison, and error normalization. Swap Postgres for MongoDB for an in-memory dict without touching this package at all.

A verify that raises an exception is treated as a 500 (server/DB error), not a 401 (bad password) — keeps your error codes honest.

Password helpers (shared so signup and login use identical hashing):

hashed = CredentialsStrategy.hash_password("plaintext")               # for signup
ok     = CredentialsStrategy.verify_password("plaintext", hashed)      # used internally by verify()

Signup itself is intentionally not abstracted — creating a user is a write to your specific schema. See server_example.py for a plain FastAPI route using CredentialsStrategy.hash_password() for consistency.

---

Adding your own strategy (e.g. GitHub)

Implement two async methods on a BaseStrategy subclass:

from auth_pipeline import BaseStrategy, AuthError, AuthResult, AuthUser
from fastapi import Request
from fastapi.responses import RedirectResponse, Response

class GitHubStrategy(BaseStrategy):
    name = "github"

    def __init__(self, *, client_id, client_secret, redirect_uri):
        self.client_id = client_id
        self.client_secret = client_secret
        self.redirect_uri = redirect_uri

    async def initiate(self, request: Request) -> Response:
        url = f"https://github.com/login/oauth/authorize?client_id={self.client_id}&redirect_uri={self.redirect_uri}"
        return RedirectResponse(url)

    async def authenticate(self, request: Request) -> AuthResult:
        code = request.query_params.get("code")
        if not code:
            raise AuthError("Missing code", status=400)

        # ... exchange code, fetch profile via GitHub's API ...

        return AuthResult(
            provider=self.name,
            user=AuthUser(id=..., email=..., name=..., picture=..., email_verified=...),
            raw={"profile": profile, "tokens": tokens},
        )

Register it and it slots into the exact same on_authenticate hook:

auth_kit.use("github", GitHubStrategy(...))

@app.get("/auth/github")
async def github_start(request: Request):
    return await auth_kit.initiate("github", request)

@app.get("/auth/github/callback")
async def github_callback(request: Request):
    return await auth_kit.complete("github", request)

Nothing else in your app needs to know GitHub exists.

---

Error handling

Every strategy should raise AuthError (not a plain Exception) for known failure cases:

raise AuthError("Invalid email or password", status=401, code="INVALID_CREDENTIALS")

Unexpected errors get auto-wrapped by AuthKit into an AuthError with status=500, code="STRATEGY_ERROR", so on_error always receives a consistent shape:

err.message   # human-readable
err.status    # HTTP status to respond with
err.code      # machine-readable, e.g. "INVALID_CREDENTIALS"
err.cause     # original exception, if any (useful for logging)

---

What this package intentionally does NOT do

• No database. CredentialsStrategy.verify and your on_authenticate hook are where persistence happens — bring your own ORM/driver (SQLAlchemy, Tortoise, motor, raw asyncpg, whatever).
• No JWT signing. on_authenticate gets a normalized user; signing tokens, setting cookies, and session management are yours to control.
• No signup flow. Creating a user is schema-specific — write that route directly, using CredentialsStrategy.hash_password() for consistency.

---

File structure

auth_pipeline/
├── core/
│   ├── auth_kit.py        ← orchestrator: initiate(), complete(), use()
│   ├── base_strategy.py   ← ABC every strategy implements
│   ├── result.py          ← AuthResult / AuthUser dataclasses
│   └── errors.py          ← AuthError exception
├── strategies/
│   ├── google.py
│   └── credentials.py
├── __init__.py            ← public exports
server_example.py          ← full working example (Google + credentials)
pyproject.toml

---

Running the example

pip install -e .
pip install uvicorn pyjwt

export GOOGLE_CLIENT_ID=your-id
export GOOGLE_CLIENT_SECRET=your-secret
export JWT_SECRET=some-long-random-string

uvicorn server_example:app --reload --port 8000

---

Differences from the JS version (and why)

JS (Express)	Python (FastAPI)	Reason
BaseStrategy is duck-typed	BaseStrategy is an ABC	Python has real interfaces — subclasses missing a method can't even be instantiated
Result is a plain object	AuthResult / AuthUser are @dataclass	Typed, autocompletable, still just data
initiate() calls res.redirect(url)	initiate() returns a RedirectResponse	FastAPI handlers return responses rather than mutating one
bcryptjs (pure JS)	bcrypt (C extension, prebuilt wheels)	Python's bcrypt ships prebuilt wheels for all major platforms, so there's no native-build pain like Node sometimes has

The wire shape (emailVerified, camelCase) is kept identical to the JS package on purpose — if your frontend talks to both a Node and a Python service, the JSON looks the same either way.

---

License

MIT