fastapi-principal 0.1.0

Identity management for FastAPI.

fastapi-principal

English | 中文说明

fastapi-principal is a small authorization toolkit for FastAPI, inspired by the API and permission model of flask-principal.

It does not authenticate users by itself. Bring your own Session, Cookie, JWT, OAuth, API Key, or database lookup, then use fastapi-principal to keep the current request identity in an async-safe context and check whether that identity has the required permissions.

Highlights

• flask-principal style Need, Identity, Permission, and Denial objects.
• Async-safe request identity storage with contextvars.
• FastAPI middleware integration through Principal(app).
• Route protection with Depends(permission.require(403)).
• Sync and async identity loaders and savers.
• Permission composition with |, &, and ~.
• Context manager and decorator support for fine-grained checks and migrations.
• Lightweight signal API compatible with (sender, identity) handlers.

Install

pip install fastapi-principal

Quick Start

from fastapi import Depends, FastAPI, Request
from fastapi_principal import Identity, Permission, Principal, RoleNeed
from fastapi_principal import get_identity, identity_loaded

app = FastAPI()
principal = Principal(app)

admin = Permission(RoleNeed("admin"))


@principal.identity_loader
async def load_identity(request: Request):
    user_id = request.headers.get("X-User-Id")
    if user_id is None:
        return None
    return Identity(user_id, auth_type="header")


@identity_loaded.connect
def add_roles(sender, identity: Identity):
    # Load roles, actions, and resource permissions from your own storage.
    if identity.id == "alice":
        identity.provides.add(RoleNeed("admin"))


@app.get("/admin", dependencies=[Depends(admin.require(403))])
async def admin_view():
    return {"message": "Hello, admin"}


@app.get("/me")
async def me():
    identity = get_identity()
    return {"id": identity.id, "auth_type": identity.auth_type}

Request examples:

curl -i http://localhost:8000/admin
curl -i -H "X-User-Id: alice" http://localhost:8000/admin

Mental Model

The library has four moving parts:

1. Need is one capability, such as RoleNeed("admin").
2. Identity is the current user or actor and owns a set of provided needs.
3. Permission describes the needs required by a resource.
4. Principal loads one identity per request and stores it in a context variable.

The request lifecycle looks like this:

1. FastAPI receives a request.
2. Principal middleware calls registered identity loaders, newest first.
3. The first loader returning an Identity wins.
4. If no loader returns an identity, AnonymousIdentity() is used.
5. identity_loaded is fired so the app can add roles and permissions.
6. Route dependencies or endpoint code call permission.require(...).
7. The identity context is reset after the response is produced.

request.state.identity is also populated for code that prefers request-local state over get_identity().

Needs

Needs are hashable named tuples. They can represent users, roles, actions, or resource-level permissions.

from fastapi_principal import ActionNeed, ItemNeed, RoleNeed, TypeNeed, UserNeed

UserNeed(42)                 # Need(method="id", value=42)
RoleNeed("admin")            # Need(method="role", value="admin")
TypeNeed("service-account")  # Need(method="type", value="service-account")
ActionNeed("publish")        # Need(method="action", value="publish")
ItemNeed("edit", 7, "post")  # ItemNeed(method="edit", value=7, type="post")

An authenticated Identity(id) automatically provides UserNeed(id). AnonymousIdentity() provides no needs.

Permissions

Permission(*needs) grants access when at least one required need is present.

editor_or_admin = Permission(RoleNeed("editor"), RoleNeed("admin"))

Denial(*needs) grants access unless one of those needs is present.

from fastapi_principal import Denial

not_banned = Denial(RoleNeed("banned"))

Permissions also keep flask-principal style set operations:

admin = Permission(RoleNeed("admin"))
editor = Permission(RoleNeed("editor"))
not_banned = Denial(RoleNeed("banned"))

admin_or_editor = admin.union(editor)
admin_only = admin_or_editor - editor
required_banned_role = not_banned.reverse()
is_subset = admin in admin_or_editor

Composition

Use Python operators to build richer rules:

admin = Permission(RoleNeed("admin"))
editor = Permission(RoleNeed("editor"))
manager = Permission(RoleNeed("manager"))
banned = Permission(RoleNeed("banned"))

admin_or_editor = admin | editor
editor_manager = editor & manager
not_banned = ~banned
policy = admin | (editor & manager & ~banned)

Permission(a, b) means "a or b". Use Permission(a) & Permission(b) when both needs are required.

FastAPI Usage

Protect a Route

@app.get("/admin", dependencies=[Depends(admin.require(403))])
async def admin_view():
    return {"ok": True}

permission.require(status_code) returns an IdentityContext that FastAPI can call directly. The explicit dependency property is also available:

Depends(admin.require(403).dependency)

If no status code is provided and access is denied, PermissionDenied is raised.

Use Inside an Endpoint

@app.post("/posts/{post_id}")
async def update_post(post_id: int):
    permission = Permission(ItemNeed("edit", post_id, "post"))
    with permission.require(403):
        return {"updated": post_id}

Use as a Decorator

@admin.require(403)
async def rebuild_index():
    return {"status": "queued"}

Both sync and async functions are supported.

Check Manually

identity = get_identity()

if identity.can(admin):
    ...

if admin.can():
    ...

admin.test(403)

Loading Identities

Register loaders with @principal.identity_loader. Loaders receive the current Request and may be sync or async.

@principal.identity_loader
def load_from_header(request: Request):
    user_id = request.headers.get("X-User-Id")
    return Identity(user_id) if user_id else None


@principal.identity_loader
async def load_from_session(request: Request):
    user_id = request.session.get("user_id")
    return Identity(user_id, auth_type="session") if user_id else None

The newest loader runs first. The first non-None identity wins. Loader exceptions are logged and the next loader is tried.

Enriching Identities

Use identity_loaded to add roles, actions, or item-level needs after an identity is loaded.

@identity_loaded.connect
def add_needs(sender, identity: Identity):
    if identity.id is None:
        return

    user = get_user_from_db(identity.id)
    for role in user.roles:
        identity.provides.add(RoleNeed(role.name))

For FastAPI-first code, a single-argument handler is also accepted:

@identity_loaded.connect
def add_default_need(identity: Identity):
    identity.provides.add(RoleNeed("member"))

Sender filtering is supported:

@identity_loaded.connect(sender=app)
def add_app_needs(sender, identity: Identity):
    ...

Persisting Identity Changes

Use Principal.set_identity() when login/logout code must persist a new identity for future requests.

@principal.identity_saver
async def save_identity(request: Request, identity: Identity):
    request.session["user_id"] = identity.id
    request.session["auth_type"] = identity.auth_type


@app.post("/login")
async def login(request: Request):
    identity = Identity("alice", auth_type="password")
    await principal.set_identity(request, identity)
    return {"status": "ok"}

Identity savers are called newest first and may be sync or async.

For flask-principal style in-request changes, use identity_changed:

from fastapi_principal import identity_changed

identity_changed.send(app, identity=Identity("alice"))

identity_changed updates the active context identity and then notifies its own receivers. It does not run identity savers because it does not receive the request object.

App Factory Pattern

principal = Principal()


def create_app():
    app = FastAPI()
    principal.init_app(app)
    return app

flask-principal Compatibility Notes

• Need, UserNeed, RoleNeed, TypeNeed, ActionNeed, and ItemNeed follow the same tuple-based model.
• Identity(id) automatically provides UserNeed(id).
• Permission, Denial, union, difference, reverse, and subset checks follow flask-principal style semantics.
• identity_loaded.connect(handler) supports (sender, identity).
• identity_changed.send(sender, identity=...) is available for in-request identity changes.
• FastAPI-specific persistence should use await principal.set_identity(request, identity).

API Reference

| Symbol | Description | | ------------------------------ | ------------------------------------------------------------------------ | ----------------------------- | | Need | namedtuple("Need", ["method", "value"]) | | ItemNeed | namedtuple("ItemNeed", ["method", "value", "type"]) | | UserNeed(v) | Shortcut for Need("id", v) | | RoleNeed(v) | Shortcut for Need("role", v) | | TypeNeed(v) | Shortcut for Need("type", v) | | ActionNeed(v) | Shortcut for Need("action", v) | | Identity(id, auth_type=None) | Active identity; adds UserNeed(id) when id is not None | | AnonymousIdentity() | Identity with no provided needs | | Permission(*needs) | Grants when any required need is present and no excluded need is present | | Denial(*needs) | Grants unless any excluded need is present | | BasePermission | Base class for custom permission types | | OrPermission | Result of p1 | p2 for non-plain permissions | | AndPermission | Result of p1 & p2 | | NotPermission | Result of ~p | | PermissionDenied | Raised when no HTTP status code is configured | | IdentityContext | Return value of permission.require() | | Principal | FastAPI middleware and identity hook manager | | get_identity() | Return the active request identity or anonymous identity | | set_identity(identity) | Change the active context identity and fire identity_loaded | | identity_loaded | Signal fired after identity loading or context identity changes | | identity_changed | Signal for flask-principal style identity changes |

License

MIT