fastapi-principal 0.1.1

Identity management for FastAPI.

fastapi-principal

English | 中文说明

fastapi-principal is a small authorization toolkit for FastAPI. It keeps the core permission model of flask-principal. The request lifecycle, loader/saver hooks, and persistence APIs are adapted for FastAPI rather than being a drop-in Flask-Principal port.

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, Permission, and Denial objects, with a FastAPI-friendly Identity model.
• 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

fastapi-principal follows Flask-Principal's authorization model, but its FastAPI integration is intentionally different from Flask's session and signal pipeline.

• Need, UserNeed, RoleNeed, TypeNeed, ActionNeed, and ItemNeed follow the same tuple-based model.
• Identity(id) automatically provides UserNeed(id). In Flask-Principal, applications often add that need explicitly in identity_loaded.
• 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, but it does not run identity savers because it has no Request.
• principal.identity_loader receives request, and principal.identity_saver receives (request, identity).
• FastAPI-specific persistence should use await principal.set_identity(request, identity).
• Flask-Principal's use_sessions, skip_static, and full Blinker signal API are not provided.

API Reference

Needs

Symbol	Description
Need	namedtuple("Need", ["method", "value"]).
ItemNeed	namedtuple("ItemNeed", ["method", "value", "type"]) for resource-level permissions.
UserNeed(value)	Shortcut for Need("id", value).
RoleNeed(value)	Shortcut for Need("role", value).
TypeNeed(value)	Shortcut for Need("type", value).
ActionNeed(value)	Shortcut for Need("action", value).

Identities

Symbol	Description
Identity(id, auth_type=None)	Active user or actor. Non-anonymous identities automatically provide UserNeed(id).
AnonymousIdentity()	Identity with id=None and no provided needs.
identity.can(permission)	Return whether the identity satisfies a permission.
get_identity()	Return the active request identity, or AnonymousIdentity() outside a request context.
set_identity(identity, *, sender=None)	Change the active context identity and fire identity_loaded.

Permissions

Symbol	Description
BasePermission	Base class for custom permission types. Supports OR, AND, and NOT composition.
Permission(*needs)	Grants when the identity provides at least one required need and no excluded need.
Denial(*needs)	Grants unless the identity provides one of the excluded needs.
OrPermission	Composite permission returned by OR composition.
AndPermission	Composite permission returned by p1 & p2.
NotPermission	Inverted permission returned by ~p.
permission.require(http_exception=None)	Return an IdentityContext usable as a dependency, decorator, or context manager.
permission.test(http_exception=None)	Raise immediately when the active identity does not satisfy the permission.
PermissionDenied	Raised when permission is denied and no HTTP status code is configured.
IdentityContext	Runtime permission check returned by permission.require().

Principal and Signals

Symbol	Description
Principal(app=None)	FastAPI middleware and identity hook manager. Pass app or call init_app(app).
principal.identity_loader(func)	Register a sync or async identity loader. The most recently registered loader runs first.
principal.identity_saver(func)	Register a sync or async identity saver used by principal.set_identity().
principal.set_identity(request, identity)	Change the active identity, update request.state.identity, and run savers.
identity_loaded	Signal fired after identity loading or context identity changes.
identity_changed	Flask-Principal style signal that sets the active identity, then notifies receivers.

License

MIT