Events & Hooks

AuthFort emits events for every authentication action. Use hooks to react to these events — send emails, log audit trails, update analytics, sync with external systems.

Registering Hooks

Decorator

@auth.on("user_created")
async def on_signup(event):
    await send_welcome_email(event.email)

@auth.on("login_failed")
async def on_failed_login(event):
    await log_failed_attempt(event.email, event.ip_address, event.reason)

@auth.on("user_created")
async def on_signup(event):
    await send_welcome_email(event.email)

@auth.on("login_failed")
async def on_failed_login(event):
    await log_failed_attempt(event.email, event.ip_address, event.reason)

Programmatic

async def on_signup(event):
    await send_welcome_email(event.email)

auth.add_hook("user_created", on_signup)

async def on_signup(event):
    await send_welcome_email(event.email)

auth.add_hook("user_created", on_signup)

Both approaches are equivalent. The decorator is syntactic sugar for add_hook().

Typed Handlers

Event types are importable from authfort.events for type-safe handlers:

from authfort import AuthFort
from authfort.events import UserCreated, LoginFailed, UserUpdated, UserDeleted, RateLimitExceeded

auth = AuthFort(database_url="...")

@auth.on("user_created")
async def on_signup(event: UserCreated):
    await send_welcome_email(event.email)

@auth.on("login_failed")
async def on_failed_login(event: LoginFailed):
    await log_failed_attempt(event.email, event.ip_address, event.reason)

@auth.on("user_updated")
async def on_profile_update(event: UserUpdated):
    await sync_profile(event.user_id, event.fields)

from authfort import AuthFort
from authfort.events import UserCreated, LoginFailed, UserUpdated, UserDeleted, RateLimitExceeded

auth = AuthFort(database_url="...")

@auth.on("user_created")
async def on_signup(event: UserCreated):
    await send_welcome_email(event.email)

@auth.on("login_failed")
async def on_failed_login(event: LoginFailed):
    await log_failed_attempt(event.email, event.ip_address, event.reason)

@auth.on("user_updated")
async def on_profile_update(event: UserUpdated):
    await sync_profile(event.user_id, event.fields)

All Events

User Lifecycle

Event	Payload Fields
user_created	user_id, email, name, provider, timestamp
user_updated	user_id, fields, timestamp
user_deleted	user_id, email, timestamp
user_banned	user_id, timestamp
user_unbanned	user_id, timestamp

Authentication

Event	Payload Fields
login	user_id, email, provider, ip_address, user_agent, timestamp
login_failed	email, reason, ip_address, user_agent, timestamp
logout	user_id, timestamp
token_refreshed	user_id, ip_address, user_agent, timestamp

Roles

Event	Payload Fields
role_added	user_id, role, timestamp
role_removed	user_id, role, timestamp

Sessions

Event	Payload Fields
session_revoked	user_id, session_id, revoke_all, timestamp

OAuth

Event	Payload Fields
oauth_link	user_id, email, provider, timestamp

Password

Event	Payload Fields
password_reset_requested	user_id, email, timestamp
password_reset	user_id, timestamp
password_changed	user_id, timestamp

Email Verification

Event	Payload Fields
email_verification_requested	user_id, email, token, timestamp
email_verified	user_id, email, timestamp

Passwordless

Event	Payload Fields
magic_link_requested	user_id, email, token, timestamp
magic_link_login	user_id, email, timestamp
email_otp_requested	user_id, email, code, timestamp
email_otp_login	user_id, email, timestamp

Keys

Event	Payload Fields
key_rotated	old_kid, new_kid, timestamp

Rate Limiting

Event	Payload Fields
rate_limit_exceeded	endpoint, ip_address, email, limit, key_type, timestamp

Examples

Audit Log

@auth.on("login")
async def audit_login(event):
    await db.insert("audit_log", {
        "action": "login",
        "user_id": event.user_id,
        "ip": event.ip_address,
        "user_agent": event.user_agent,
        "timestamp": event.timestamp,
    })

@auth.on("login")
async def audit_login(event):
    await db.insert("audit_log", {
        "action": "login",
        "user_id": event.user_id,
        "ip": event.ip_address,
        "user_agent": event.user_agent,
        "timestamp": event.timestamp,
    })

Rate Limiting Failed Logins

from collections import defaultdict

failed_attempts = defaultdict(int)

@auth.on("login_failed")
async def track_failures(event):
    failed_attempts[event.email] += 1
    if failed_attempts[event.email] >= 5:
        await notify_admin(f"Multiple failed logins for {event.email}")

from collections import defaultdict

failed_attempts = defaultdict(int)

@auth.on("login_failed")
async def track_failures(event):
    failed_attempts[event.email] += 1
    if failed_attempts[event.email] >= 5:
        await notify_admin(f"Multiple failed logins for {event.email}")

Welcome Email

@auth.on("user_created")
async def welcome(event):
    if event.provider == "email":
        await send_welcome_email(event.email, event.name)

@auth.on("user_created")
async def welcome(event):
    if event.provider == "email":
        await send_welcome_email(event.email, event.name)

Magic Link Delivery

@auth.on("magic_link_requested")
async def send_magic_link(event):
    await send_email(
        to=event.email,
        subject="Your login link",
        body=f"https://myapp.com/auth/magic?token={event.token}",
    )

@auth.on("magic_link_requested")
async def send_magic_link(event):
    await send_email(
        to=event.email,
        subject="Your login link",
        body=f"https://myapp.com/auth/magic?token={event.token}",
    )

Multiple Hooks

You can register multiple hooks for the same event. They run concurrently:

@auth.on("login")
async def hook_1(event):
    await update_last_login(event.user_id)

@auth.on("login")
async def hook_2(event):
    await send_login_notification(event.email)

@auth.on("login")
async def hook_1(event):
    await update_last_login(event.user_id)

@auth.on("login")
async def hook_2(event):
    await send_login_notification(event.email)