Decoupled User Management in Python

# app/main.py
"""
UserHarbor Demo – FastAPI application entry point.

Stack
─────
* userharbor         – core user-management domain logic
* userharbor-sqlalchemy – SQLite-backed UserStore
* userharbor-fastapi – FastAPI router adapter with bearer-token auth
* ConsoleEmailSender / SMTPEmailSender – from app/email.py

Environment variables (copy .env.example → .env):
  SECRET_KEY     – signing key for session tokens
  DATABASE_URL   – SQLAlchemy URL          (default: sqlite:///./users.db)
  SMTP_*         – optional SMTP credentials; falls back to console logging
"""
from __future__ import annotations

import os
from pathlib import Path

from dotenv import load_dotenv
from fastapi import FastAPI
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from userharbor import UserHarbor
from userharbor_fastapi import UserHarborFastAPI
from userharbor_sqlalchemy import SQLAlchemyUserStore

from app.email import get_email_sender
from app.routes import build_demo_router

FRONTEND_DIR = Path(__file__).parent.parent / "frontend"

load_dotenv()

# ── Database ───────────────────────────────────────────────────────────────
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./users.db")
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)

# ── UserHarbor setup ───────────────────────────────────────────────────────
store = SQLAlchemyUserStore(SessionLocal)
store.metadata.create_all(engine)  # create tables if they don't exist

harbor = UserHarbor(
    secret_key=os.getenv("SECRET_KEY", "demo-secret-key-change-in-production"),
    store=store,
    email_sender=get_email_sender(),
)

# Pre-seed roles and permissions so the demo endpoints work out of the box.
try:
    harbor.roles.create("admin")
    harbor.roles.create("editor")
    harbor.permissions.create("articles.write")
    harbor.permissions.create("users.manage")
    harbor.roles.grant_permission("admin", "articles.write")
    harbor.roles.grant_permission("admin", "users.manage")
    harbor.roles.grant_permission("editor", "articles.write")
except Exception:
    pass  # roles/permissions already exist from a previous run

# ── FastAPI adapter ────────────────────────────────────────────────────────
auth = UserHarborFastAPI(harbor)

# ── Application ────────────────────────────────────────────────────────────
app = FastAPI(
    title="UserHarbor Demo",
    description=(
        "End-to-end demonstration of UserHarbor with FastAPI + SQLAlchemy + SMTP.\n\n"
        "Workflow:\n"
        "1. `POST /auth/register` – create an account\n"
        "2. `POST /auth/verify-email` – verify your email with the token printed to console\n"
        "3. `POST /auth/login` – obtain a bearer token\n"
        "4. Use the **Authorize** button (🔒) to paste the token\n"
        "5. Try the protected `/demo/*` endpoints\n"
    ),
    version="1.0.0",
)

# Built-in auth routes: /auth/register, /auth/login, /auth/me, etc.
app.include_router(auth.router, prefix="/auth", tags=["Authentication"])

# Demo application routes that showcase role/permission guards.
app.include_router(
    build_demo_router(auth, harbor),
    prefix="/demo",
    tags=["Demo – Protected Routes"],
)

# Serve the SPA frontend
app.mount("/static", StaticFiles(directory=str(FRONTEND_DIR)), name="static")

@app.get("/", tags=["Root"], include_in_schema=False)
def serve_ui():
    """Serve the SPA shell."""
    return FileResponse(str(FRONTEND_DIR / "index.html"))

if __name__ == "__main__":
    import uvicorn

    uvicorn.run("app.main:app", host="0.0.0.0", port=8080, reload=True)

"""
Email sender implementations for the demo.

* ConsoleEmailSender – prints tokens to stdout; ideal for development when no
  SMTP server is available.
* get_email_sender()  – factory that returns an SMTPEmailSender when SMTP
  credentials are present, otherwise falls back to the console sender.
"""
from __future__ import annotations

import os

class ConsoleEmailSender:
    """Prints email content to the console instead of sending real mail."""

    def send_verification(
        self,
        username: str,
        email: str,
        verification_token: str,
    ) -> None:
        print(
            f"\n[EMAIL] Verification token for {username} ({email}):\n"
            f"  Token: {verification_token}\n"
        )

    def send_password_reset(
        self,
        username: str,
        email: str,
        reset_token: str,
    ) -> None:
        print(
            f"\n[EMAIL] Password-reset token for {username} ({email}):\n"
            f"  Token: {reset_token}\n"
        )

    def send_email_verified(self, username: str, email: str) -> None:
        print(f"\n[EMAIL] Email verified – {username} ({email})\n")

    def send_password_changed(self, username: str, email: str) -> None:
        print(f"\n[EMAIL] Password changed – {username} ({email})\n")

    def send_account_deleted(self, username: str, email: str) -> None:
        print(f"\n[EMAIL] Account deleted – {username} ({email})\n")

def get_email_sender():
    """Return an SMTPEmailSender when credentials are available, else console."""
    host = os.getenv("SMTP_HOST", "")
    username = os.getenv("SMTP_USERNAME", "")
    password = os.getenv("SMTP_PASSWORD", "")
    from_email = os.getenv("SMTP_FROM_EMAIL", "")

    if host and username and password and from_email:
        from userharbor_smtp import SMTPEmailSender  # type: ignore

        return SMTPEmailSender(
            host=host,
            port=int(os.getenv("SMTP_PORT", "587")),
            username=username,
            password=password,
            from_email=from_email,
            from_name=os.getenv("SMTP_FROM_NAME", "UserHarbor Demo"),
        )

    print("[INFO] No SMTP credentials found – using ConsoleEmailSender.")
    return ConsoleEmailSender()

"""
Demo application routes that showcase every UserHarbor feature:

Route                     Auth required    Extra guard
─────────────────────────────────────────────────────
GET  /demo/public          none            –
GET  /demo/profile         session token   –
GET  /demo/optional        optional        –
POST /demo/assign-admin    session token   – (grant admin role to a user)
GET  /demo/articles/write  session token   permission: articles.write
GET  /demo/admin/users     session token   role: admin
"""
from __future__ import annotations

from fastapi import APIRouter, Depends
from pydantic import BaseModel
from userharbor import UserHarbor
from userharbor_fastapi import UserHarborFastAPI

# ── Request / response models ──────────────────────────────────────────────

class AssignRoleRequest(BaseModel):
    username: str
    role: str

# ── Router factory ─────────────────────────────────────────────────────────

def build_demo_router(auth: UserHarborFastAPI, harbor: UserHarbor) -> APIRouter:
    router = APIRouter()

    # ── Public route – no authentication ──────────────────────────────────
    @router.get("/public")
    def public():
        """Anyone can call this endpoint – no token required."""
        return {
            "message": "This is a public endpoint. No authentication needed.",
            "hint": "Call POST /auth/register to create an account, then POST /auth/login.",
        }

    # ── Optional auth – behaves differently for guests vs. logged-in users ─
    @router.get("/optional")
    def optional(user=Depends(auth.optional_user)):
        """Shows personalised content when a valid token is present."""
        if user is None:
            return {"authenticated": False, "message": "Hello, anonymous visitor!"}
        return {
            "authenticated": True,
            "message": f"Hello, {user.username}! You are verified: {user.verified}",
        }

    # ── Protected – requires a valid session token ─────────────────────────
    @router.get("/profile")
    def profile(user=Depends(auth.current_user)):
        """Requires a valid session token (Authorization: Bearer <token>)."""
        return {
            "username": user.username,
            "email": user.email,
            "verified": user.verified,
        }

    # ── Assign a role to any user (admin can do this) ──────────────────────
    @router.post("/assign-role")
    def assign_role(
        body: AssignRoleRequest,
        user=Depends(auth.current_user),
    ):
        """
        Grant a role to a user. Requires a valid session.

        Useful for promoting someone to 'admin' or 'editor' so you can
        test the role-restricted endpoints below.
        """
        harbor.grant_role(body.username, body.role)
        return {
            "message": f"Role '{body.role}' granted to '{body.username}'.",
            "granted_by": user.username,
        }

    # ── Permission-restricted route ────────────────────────────────────────
    @router.get("/articles/write")
    def write_articles(user=Depends(auth.require_permission("articles.write"))):
        """
        Requires the **articles.write** permission.

        Roles that have this permission: `admin`, `editor`.
        Grant one with POST /demo/assign-role.
        """
        return {
            "message": f"Welcome, {user.username}! You can write articles.",
            "permission": "articles.write",
        }

    # ── Role-restricted route ──────────────────────────────────────────────
    @router.get("/admin/users")
    def admin_users(user=Depends(auth.require_role("admin"))):
        """
        Requires the **admin** role.

        Grant with: POST /demo/assign-role  { "username": "...", "role": "admin" }
        """
        return {
            "message": f"Welcome, admin {user.username}!",
            "note": "In a real app you would return a paginated user list here.",
        }

    return router

| Feature                 | UserHarbor API                                               |
| ----------------------- | ------------------------------------------------------------ |
| User registration       | `harbor.register()`                                          |
| Email verification      | `harbor.verify_email()`                                      |
| Login / bearer token    | `harbor.login()`                                             |
| Session validation      | `harbor.verify_session()`                                    |
| Current user            | `harbor.get_current_user()`                                  |
| Optional authentication | `auth.optional_user` dependency                              |
| Role management         | `harbor.roles.create()` / `harbor.grant_role()`              |
| Permission management   | `harbor.permissions.create()` / `harbor.roles.grant_permission()` |
| Role guard              | `auth.require_role("admin")` dependency                      |
| Permission guard        | `auth.require_permission("articles.write")` dependency       |
| Password change         | `harbor.change_password()`                                   |
| Password reset flow     | `harbor.send_password_reset()` / `harbor.reset_password()`   |
| Logout                  | `harbor.logout()`                                            |
| Logout all sessions     | `harbor.logout_all()`                                        |
| Account deletion        | `harbor.delete_account()`                                    |

---