Added authentication forms and routes

This commit is contained in:
2025-10-27 22:19:34 +01:00
parent e12e4389b6
commit b21161a273
27 changed files with 3336 additions and 337 deletions

View File

@@ -0,0 +1,283 @@
# FastHTML + FastAPI Authentication Integration
This package provides a complete authentication system integrating FastHTML (frontend) with FastAPI (backend).
## Architecture
```
auth/
├── __init__.py # Package initialization
├── utils.py # JWT helpers + Beforeware
├── pages/
│ ├── __init__.py
│ ├── LoginPage.py # Login form component
│ ├── RegisterPage.py # Registration form component
│ └── WelcomePage.py # Protected welcome page
└── routes.py # All routes (public + protected)
```
## Features
-**Login page** with email/password authentication
-**Registration page** with password confirmation
-**Automatic token refresh** (proactive refresh at 5 minutes before expiry)
-**Route protection** with Beforeware (whitelist support)
-**Session management** using FastHTML signed cookies
-**DaisyUI 5** components for beautiful UI
-**HTMX integration** for dynamic form submissions
## Dependencies
Install the required packages:
```bash
pip install fasthtml httpx python-jose[cryptography] --break-system-packages
```
Note: `python-jose` may already be installed if you have FastAPI.
## Configuration
### 1. Update API configuration in `auth/utils.py`
```python
API_BASE_URL = "http://localhost:5001" # Your FastAPI backend URL
JWT_SECRET = "jwt-secret-to-change" # Must match your FastAPI secret
```
### 2. Create your `main.py`
```python
from fasthtml.common import fast_app, serve, Link, Script
from auth import create_auth_beforeware, setup_auth_routes
# Import your existing FastAPI auth app
# from your_auth_module import create_auth_app_for_sqlite
# Create Beforeware for route protection
beforeware = create_auth_beforeware()
# Create FastHTML app
app, rt = fast_app(
before=beforeware,
hdrs=(
# DaisyUI 5 CSS
Link(rel='stylesheet', href='https://cdn.jsdelivr.net/npm/daisyui@5/daisyui.css'),
# Tailwind CSS 4
Script(src='https://cdn.jsdelivr.net/npm/@tailwindcss/browser@4'),
)
)
# Mount FastAPI auth backend
# auth_api = create_auth_app_for_sqlite("Users.db", "jwt-secret-to-change")
# app.mount("/auth", auth_api)
# Setup authentication routes
setup_auth_routes(app, rt)
if __name__ == "__main__":
serve()
```
## Usage
### Starting the application
```bash
python main.py
```
The application will be available at `http://localhost:5001`.
### Routes
**Public routes:**
- `GET /login` - Display login page
- `POST /login` - Handle login submission
- `GET /register` - Display registration page
- `POST /register` - Handle registration submission
**Protected routes (require authentication):**
- `GET /` - Welcome page (home)
- `GET /welcome` - Alternative welcome page
- `POST /logout` - Logout and revoke tokens
### Adding custom protected routes
```python
@rt("/dashboard")
def get_dashboard(session, auth):
"""
Custom protected route.
The 'auth' parameter is automatically injected by the Beforeware.
"""
user_info = session.get('user_info', {})
return Div(
H1(f"Dashboard for {user_info.get('email', 'User')}"),
# Your dashboard content
)
```
### Extending the whitelist
If you need to add more public routes:
```python
beforeware = create_auth_beforeware(
additional_patterns=[
'/about',
'/contact',
r'/api/.*', # Regex patterns are supported
]
)
```
## How it works
### 1. Beforeware Protection
The `auth_before()` function runs before every route (except whitelisted ones):
1. Checks if `access_token` exists in session
2. Validates token expiration
3. If token expires in < 5 minutes, automatically refreshes it
4. If refresh fails or no token exists, redirects to `/login`
### 2. Token Refresh Flow
```
User Request → Beforeware checks token
Token expires in < 5 min?
↓ Yes
Call /auth/refresh with refresh_token
Update session with new tokens
Continue to route handler
```
### 3. Session Storage
Session contains:
- `access_token` (JWT, 30 minutes validity)
- `refresh_token` (Opaque, 7 days validity)
- `user_info` (email, id, etc.)
## Customization
### Changing the refresh threshold
Edit `auth/utils.py`:
```python
TOKEN_REFRESH_THRESHOLD_MINUTES = 5 # Change to your preferred value
```
### Customizing page components
Each page component is a class with:
- `__init__()` - Constructor with parameters
- `_render()` - HTML generation logic
- `__ft__()` - FastHTML protocol method
Example:
```python
class LoginPage:
def __init__(self, error_message=None, redirect_url="/"):
self.error_message = error_message
self.redirect_url = redirect_url
def _render(self):
# Build your custom HTML here
return Div(...)
def __ft__(self):
return self._render()
```
### Styling with DaisyUI 5
DaisyUI 5 provides semantic class names:
```python
# Button
Button("Click me", cls="btn btn-primary")
# Input
Input(type="email", cls="input input-bordered")
# Card
Div(
Div(
H2("Title", cls="card-title"),
P("Content"),
cls="card-body"
),
cls="card bg-base-100 shadow-xl"
)
# Alert
Div(
Span("Success message"),
cls="alert alert-success"
)
```
## Troubleshooting
### Issue: "Module not found: auth"
Make sure the `auth` directory is in your Python path or in the same directory as `main.py`.
### Issue: "Connection refused" when calling API
Check that:
1. `API_BASE_URL` in `auth/utils.py` matches your FastAPI server
2. Your FastAPI backend is running
3. The `/auth` endpoints are correctly mounted
### Issue: Token refresh not working
Verify:
1. `JWT_SECRET` in `auth/utils.py` matches your FastAPI secret
2. Your FastAPI `/auth/refresh` endpoint is working
3. The refresh token is being stored correctly in the session
### Issue: Infinite redirect loop
Check:
1. The `/login` route is in the whitelist (it should be by default)
2. Your FastAPI `/auth/login` endpoint returns `access_token` and `refresh_token`
## Testing
To test the authentication flow:
1. Start your application
2. Navigate to `http://localhost:5001/`
3. You should be redirected to `/login`
4. Register a new account at `/register`
5. Login with your credentials
6. You should see the welcome page
7. Your session will automatically refresh tokens as needed
## API Integration
This package expects your FastAPI backend to provide these endpoints:
- `POST /auth/login` - Returns `access_token` and `refresh_token`
- `POST /auth/register` - Creates a new user
- `POST /auth/refresh` - Refreshes the access token
- `GET /auth/me` - Returns current user info (requires Bearer token)
- `POST /auth/logout` - Revokes refresh token
## License
This code is provided as-is for educational and development purposes.
## Questions?
Feel free to modify and extend this authentication system according to your needs!

View File

View File

@@ -0,0 +1,88 @@
from fasthtml.components import *
class LoginPage:
def __init__(self, error_message=None, success_message=None):
self.error_message = error_message
self.success_message = success_message
def render(self):
message_alert = None
if self.error_message:
message_alert = Div(
P(self.error_message, cls="text-sm"),
cls="bg-error border border-red-400 text-red-700 px-4 py-3 rounded mb-4"
)
elif self.success_message:
message_alert = Div(
P(self.success_message, cls="text-sm"),
cls="bg-success border border-green-400 text-green-700 px-4 py-3 rounded mb-4"
)
return Div(
# Page title
H1("Sign In", cls="text-3xl font-bold text-center mb-6"),
# Login Form
Div(
# Message alert
message_alert if message_alert else "",
# Email login form
Form(
# Email field
Div(
Label("Email", For="email", cls="block text-sm font-medium text-gray-700 mb-1"),
Input(
type="email",
id="email",
name="email",
placeholder="you@example.com",
required=True,
cls="w-full px-3 py-2 border rounded-md focus:outline-none focus:ring-2"
),
cls="mb-4"
),
# Password field
Div(
Label("Password", For="password", cls="block text-sm font-medium text-gray-700 mb-1"),
Input(
type="password",
id="password",
name="password",
placeholder="Your password",
required=True,
cls="w-full px-3 py-2 border rounded-md focus:outline-none focus:ring-2"
),
cls="mb-6"
),
# Submit button
Button(
"Sign In",
type="submit",
cls="btn w-full font-bold py-2 px-4 rounded"
),
action="/login-p",
method="post",
cls="mb-6"
),
# Registration link
Div(
P(
"Don't have an account? ",
A("Register here", href="/register", cls="text-blue-600 hover:underline"),
cls="text-sm text-gray-600 text-center"
)
),
cls="p-8 rounded-lg shadow-2xl max-w-md mx-auto"
)
)
def __ft__(self):
return self.render()

View File

@@ -0,0 +1,125 @@
from fasthtml.components import *
class RegisterPage:
def __init__(self, error_message: str = None):
self.error_message = error_message
def register_page(self, error_message: str):
self.error_message = error_message
return self.__ft__()
def __ft__(self):
"""
Create the registration page.
Args:
error_message: Optional error message to display
Returns:
Components representing the registration page
"""
# Create alert for error message
error_alert = None
if self.error_message:
error_alert = Div(
P(self.error_message, cls="text-sm"),
cls="bg-soft bg-error border border-red-400 text-red-700 px-4 py-3 rounded mb-4"
)
return Div(
# Page title
H1("Create an Account", cls="text-3xl font-bold text-center mb-6"),
# Registration Form
Div(
# Error alert
error_alert if error_alert else "",
Form(
# Username field
Div(
Label("Username", For="username", cls="block text-sm font-medium text-gray-700 mb-1"),
Input(
type="text",
id="username",
name="username",
placeholder="Choose a username",
required=True,
minlength=3,
maxlength=30,
pattern="[a-zA-Z0-9_-]+",
cls="w-full px-3 py-2 border rounded-md focus:outline-none focus:ring-2"
),
P("Only letters, numbers, underscores, and hyphens", cls="text-xs text-gray-500 mt-1"),
cls="mb-4"
),
# Email field
Div(
Label("Email", For="email", cls="block text-sm font-medium text-gray-700 mb-1"),
Input(
type="email",
id="email",
name="email",
placeholder="you@example.com",
required=True,
cls="w-full px-3 py-2 border rounded-md focus:outline-none focus:ring-2"
),
cls="mb-4"
),
# Password field
Div(
Label("Password", For="password", cls="block text-sm font-medium text-gray-700 mb-1"),
Input(
type="password",
id="password",
name="password",
placeholder="Create a password",
required=True,
minlength=8,
cls="w-full px-3 py-2 border rounded-md focus:outline-none focus:ring-2"
),
P("At least 8 characters with uppercase, lowercase, and number", cls="text-xs text-gray-500 mt-1"),
cls="mb-4"
),
# Confirm password field
Div(
Label("Confirm Password", For="confirm_password", cls="block text-sm font-medium text-gray-700 mb-1"),
Input(
type="password",
id="confirm_password",
name="confirm_password",
placeholder="Confirm your password",
required=True,
cls="w-full px-3 py-2 border rounded-md focus:outline-none focus:ring-2"
),
cls="mb-6"
),
# Submit button
Button(
"Create Account",
type="submit",
cls="btn w-full font-bold py-2 px-4 rounded"
),
# Registration link
Div(
P(
"Already have an account? ",
A("Sign in here", href="/login", cls="text-blue-600 hover:underline"),
cls="text-sm text-gray-600 text-center"
)
),
action="register-p",
method="post",
cls="mb-6"
),
cls="p-8 rounded-lg shadow-2xl max-w-md mx-auto"
)
)

View File

@@ -0,0 +1,216 @@
"""
Register page component for FastHTML application.
Provides a styled registration form using DaisyUI 5 components.
"""
from fasthtml.common import Div, Form, Input, Button, H1, P, A, Label
class RegisterPage:
"""
Register page component with email and password fields.
Attributes:
error_message: Optional error message to display
redirect_url: URL to redirect after successful registration
"""
def __init__(self, error_message: str = None, redirect_url: str = "/login"):
"""
Initialize register page.
Args:
error_message: Optional error message to display
redirect_url: URL to redirect after successful registration (default: "/login")
"""
self.error_message = error_message
self.redirect_url = redirect_url
def _render(self):
"""
Render the register page HTML structure.
Returns:
FastHTML component tree for the register page
"""
# Error alert component (only shown if error_message exists)
error_alert = None
if self.error_message:
error_alert = Div(
Div(
self._svg_icon(),
self.error_message,
cls="flex items-center gap-2"
),
role="alert",
cls="alert alert-error mb-4",
id="error-alert"
)
# Success message container (empty by default, filled by HTMX on success)
success_alert = Div(id="success-alert")
# Register form
register_form = Form(
# Hidden field for redirect URL
Input(type="hidden", name="redirect_url", value=self.redirect_url),
# Email input
Label(
Div("Email", cls="label-text"),
cls="form-control w-full mb-4"
)(
Input(
type="email",
name="email",
placeholder="[email protected]",
required=True,
cls="input input-bordered w-full"
)
),
# Password input
Label(
Div("Password", cls="label-text"),
Div("Must be at least 8 characters", cls="label-text-alt text-base-content/60"),
cls="form-control w-full mb-4"
)(
Input(
type="password",
name="password",
placeholder="Create a password",
required=True,
minlength="8",
cls="input input-bordered w-full"
)
),
# Confirm password input
Label(
Div("Confirm Password", cls="label-text"),
cls="form-control w-full mb-4"
)(
Input(
type="password",
name="confirm_password",
placeholder="Confirm your password",
required=True,
minlength="8",
cls="input input-bordered w-full"
)
),
# Submit button
Button(
"Create Account",
type="submit",
cls="btn btn-primary w-full"
),
method="post",
action="/register",
hx_post="/register",
hx_target="#register-card",
hx_swap="outerHTML"
)
# Main card container
card = Div(
Div(
# Card title
H1("Create Account", cls="text-3xl font-bold text-center mb-2"),
P("Sign up for a new account", cls="text-center text-base-content/70 mb-6"),
# Success alert
success_alert,
# Error alert
error_alert if error_alert else Div(),
# Register form
register_form,
# Login link
Div(
P(
"Already have an account? ",
A("Sign in", href="/login", cls="link link-primary"),
cls="text-center text-sm"
),
cls="mt-4"
),
cls="card-body"
),
cls="card bg-base-100 w-96 shadow-xl",
id="register-card"
)
# Full page container
return Div(
card,
cls="flex items-center justify-center min-h-screen bg-base-200"
)
def _svg_icon(self):
"""
Helper method to create error icon SVG.
Returns:
SVG icon as NotStr
"""
from fasthtml.common import NotStr
return NotStr('''
<svg xmlns="http://www.w3.org/2000/svg"
class="h-6 w-6 shrink-0 stroke-current"
fill="none"
viewBox="0 0 24 24">
<path stroke-linecap="round"
stroke-linejoin="round"
stroke-width="2"
d="M10 14l2-2m0 0l2-2m-2 2l-2-2m2 2l2 2m7-2a9 9 0 11-18 0 9 9 0 0118 0z" />
</svg>
''')
def __ft__(self):
"""
FastHTML rendering protocol method.
Returns:
Rendered register page component
"""
return self._render()
def create_success_message(message: str = "Account created successfully!"):
"""
Create a success alert component for HTMX response.
Args:
message: Success message to display
Returns:
Success alert component
"""
from fasthtml.common import NotStr
return Div(
Div(
NotStr('''
<svg xmlns="http://www.w3.org/2000/svg"
class="h-6 w-6 shrink-0 stroke-current"
fill="none"
viewBox="0 0 24 24">
<path stroke-linecap="round"
stroke-linejoin="round"
stroke-width="2"
d="M9 12l2 2 4-4m6 2a9 9 0 11-18 0 9 9 0 0118 0z" />
</svg>
'''),
message,
cls="flex items-center gap-2"
),
role="alert",
cls="alert alert-success mb-4"
)

View File

@@ -0,0 +1,185 @@
"""
Welcome page component for FastHTML application.
Provides a protected welcome page that displays user information.
This page is only accessible to authenticated users.
"""
from fasthtml.common import Div, H1, H2, P, Button, A
from typing import Optional, Dict, Any
class WelcomePage:
"""
Welcome page component for authenticated users.
Attributes:
user_info: Dictionary containing user information (email, id, etc.)
redirect_url: URL for logout redirect
"""
def __init__(self, user_info: Optional[Dict[str, Any]] = None, redirect_url: str = "/login"):
"""
Initialize welcome page.
Args:
user_info: Dictionary containing user information
redirect_url: URL to redirect after logout (default: "/login")
"""
self.user_info = user_info or {}
self.redirect_url = redirect_url
def _render(self):
"""
Render the welcome page HTML structure.
Returns:
FastHTML component tree for the welcome page
"""
# Extract user info
email = self.user_info.get('email', 'Guest')
user_id = self.user_info.get('id', 'N/A')
# Hero section
hero = Div(
Div(
# Welcome message
H1(
f"Welcome back, {email}!",
cls="text-5xl font-bold"
),
P(
"You have successfully logged in to your account.",
cls="py-6 text-lg"
),
# Action buttons
Div(
Button(
"Logout",
cls="btn btn-primary",
hx_post="/logout",
hx_target="body",
hx_swap="outerHTML"
),
cls="flex gap-2"
),
cls="max-w-md"
),
cls="hero min-h-screen bg-base-200"
)
# User info card
info_card = Div(
Div(
H2("Your Information", cls="card-title mb-4"),
# User details
Div(
self._info_row("Email", email),
self._info_row("User ID", str(user_id)),
cls="space-y-2"
),
cls="card-body"
),
cls="card bg-base-100 shadow-xl w-full max-w-md"
)
# Main container
return Div(
# Hero section
hero,
# Info card (positioned absolutely over hero)
Div(
info_card,
cls="absolute top-20 right-20 hidden lg:block"
),
cls="relative"
)
def _info_row(self, label: str, value: str):
"""
Create an information row for displaying user details.
Args:
label: Label for the information
value: Value to display
Returns:
Div containing label and value
"""
return Div(
Div(
P(label, cls="font-semibold text-sm text-base-content/70"),
P(value, cls="text-base"),
cls="flex flex-col"
),
cls="py-2 border-b border-base-300 last:border-b-0"
)
def __ft__(self):
"""
FastHTML rendering protocol method.
Returns:
Rendered welcome page component
"""
return self._render()
class SimpleWelcomePage:
"""
Simpler version of welcome page without user info card.
Useful for quick prototyping or minimal designs.
"""
def __init__(self, user_email: str = "Guest"):
"""
Initialize simple welcome page.
Args:
user_email: User's email address
"""
self.user_email = user_email
def _render(self):
"""
Render the simple welcome page.
Returns:
FastHTML component tree
"""
return Div(
Div(
Div(
H1(f"Welcome, {self.user_email}!", cls="text-4xl font-bold mb-4"),
P("You are now logged in.", cls="mb-6 text-lg"),
Button(
"Logout",
cls="btn btn-primary",
hx_post="/logout",
hx_target="body",
hx_swap="outerHTML"
),
cls="text-center"
),
cls="hero-content"
),
cls="hero min-h-screen bg-base-200"
)
def __ft__(self):
"""
FastHTML rendering protocol method.
Returns:
Rendered simple welcome page
"""
return self._render()

View File

View File

@@ -0,0 +1,195 @@
"""
Authentication routes for FastHTML application.
Provides routes for login, register, logout, and protected pages.
"""
from fasthtml.common import RedirectResponse
from myauth import create_auth_app_for_sqlite
from ..auth.pages.LoginPage import LoginPage
from ..auth.pages.RegisterPage import RegisterPage
from ..auth.pages.WelcomePage import WelcomePage
from ..auth.utils import (
login_user,
register_user,
logout_user,
get_user_info
)
def setup_auth_routes(app, rt, mount_auth_app=True, sqlite_db_path="Users.db"):
"""
Setup all authentication and protected routes.
Args:
app: FastHTML application instance
rt: Route decorator from FastHTML
mount_auth_app: Whether to mount the auth FastApi API routes
sqlite_db_path: by default, create a new SQLite database at this path
"""
# ============================================================================
# PUBLIC ROUTES (Login & Register)
# ============================================================================
@rt("/login")
def get(error: str = None):
"""
Display login page.
Args:
error: Optional error message from query params
Returns:
LoginPage component
"""
return LoginPage(error_message=error)
@rt("/login-p")
def post(email: str, password: str, session, redirect_url: str = "/"):
"""
Handle login form submission.
Args:
email: User email from form
password: User password from form
session: FastHTML session object
redirect_url: URL to redirect after successful login
Returns:
RedirectResponse on success, or LoginPage with error on failure
"""
# Attempt login
auth_data = login_user(email, password)
if auth_data:
# Login successful - store tokens in session
session['access_token'] = auth_data['access_token']
session['refresh_token'] = auth_data['refresh_token']
# Get user info and store in session
user_info = get_user_info(auth_data['access_token'])
if user_info:
session['user_info'] = user_info
# Redirect to protected page
return RedirectResponse(redirect_url, status_code=303)
else:
# Login failed - return error message via HTMX
return LoginPage(error_message="Invalid email or password. Please try again.")
@rt("/register")
def get(error: str = None):
"""
Display registration page.
Args:
error: Optional error message from query params
Returns:
RegisterPage component
"""
return RegisterPage(error_message=error)
@rt("/register-p")
def post(email: str, username: str, password: str, confirm_password: str, session):
"""
Handle registration form submission.
Args:
email: User email from form
username: User name of the
password: User password from form
confirm_password: Password confirmation from form
session: FastHTML session object
Returns:
RegisterPage with success/error message via HTMX
"""
# Validate password confirmation
if password != confirm_password:
return RegisterPage(error_message="Passwords do not match. Please try again.")
# Validate password length
if len(password) < 8:
return RegisterPage(error_message="Password must be at least 8 characters long.")
# Attempt registration
result = register_user(email, username, password)
if result:
# Registration successful - show success message and auto-login
auth_data = login_user(email, password)
if auth_data:
# Store tokens in session
session['access_token'] = auth_data['access_token']
session['refresh_token'] = auth_data['refresh_token']
# Get user info and store in session
user_info = get_user_info(auth_data['access_token'])
if user_info:
session['user_info'] = user_info
# Redirect to welcome page
return RedirectResponse("/", status_code=303)
else:
# Auto-login failed, redirect to login page
return RedirectResponse("/login", status_code=303)
else:
# Registration failed
return RegisterPage(error_message="Registration failed. Email may already be in use.")
# ============================================================================
# PROTECTED ROUTES (Require authentication)
# ============================================================================
@rt("/welcome")
def get(session, auth):
"""
Alternative welcome page route (protected).
Args:
session: FastHTML session object
auth: User auth info from request scope
Returns:
WelcomePage component
"""
user_info = session.get('user_info', {})
return WelcomePage(user_info=user_info)
@rt("/logout")
def post(session):
"""
Handle logout request.
Revokes refresh token and clears session.
Args:
session: FastHTML session object
Returns:
RedirectResponse to login page
"""
# Get refresh token from session
refresh_token = session.get('refresh_token')
# Revoke refresh token on backend
if refresh_token:
logout_user(refresh_token)
# Clear session
session.clear()
# Redirect to login page
return RedirectResponse("/login", status_code=303)
def mount_auth_fastapi_api():
# Mount FastAPI auth backend
auth_api = create_auth_app_for_sqlite(sqlite_db_path, "jwt-secret-to-change")
app.mount("/auth", auth_api)
if mount_auth_app:
mount_auth_fastapi_api()

View File

@@ -0,0 +1,302 @@
"""
Authentication utilities for FastHTML application.
This module provides:
- JWT token validation and refresh logic
- Beforeware for protecting routes
- Helper functions for API communication
"""
from datetime import datetime, timezone
from typing import Optional, Dict, Any, List
import httpx
from fasthtml.common import RedirectResponse, Beforeware
from jose import jwt, JWTError
# Configuration
API_BASE_URL = "http://localhost:5001" # Base URL for FastAPI backend
JWT_SECRET = "jwt-secret-to-change" # Must match FastAPI secret
JWT_ALGORITHM = "HS256"
TOKEN_REFRESH_THRESHOLD_MINUTES = 5 # Refresh token if expires in less than 5 minutes
# Default patterns to skip authentication
DEFAULT_SKIP_PATTERNS = [
r'/favicon\.ico',
r'/static/.*',
r'.*\.css',
r'.*\.js',
'/login',
'/login-p',
'/register',
'/register-p',
'/logout',
]
http_client = httpx
def create_auth_beforeware(additional_patterns: Optional[List[str]] = None) -> Beforeware:
"""
Create a Beforeware instance for route protection.
Args:
additional_patterns: Optional list of additional URL patterns to skip authentication
Returns:
Beforeware instance configured for authentication
"""
patterns = DEFAULT_SKIP_PATTERNS.copy()
if additional_patterns:
patterns.extend(additional_patterns)
return Beforeware(auth_before, skip=patterns)
def auth_before(request, session):
"""
Beforeware function that runs before each protected route.
Checks authentication status and automatically refreshes tokens if needed.
Redirects to login if authentication fails.
Args:
request: Starlette request object
session: FastHTML session object
Returns:
RedirectResponse to login page if authentication fails, None otherwise
"""
# Get tokens from session
access_token = session.get('access_token')
refresh_token = session.get('refresh_token')
print(f"path={request.scope['path']}, {session=}, {access_token=}, {refresh_token=}")
# If no access token, redirect to login
if not access_token:
return RedirectResponse('/login', status_code=303)
# Validate access token and check expiration
try:
payload = decode_jwt(access_token)
exp_timestamp = payload.get('exp')
if exp_timestamp:
exp_datetime = datetime.fromtimestamp(exp_timestamp, tz=timezone.utc)
now = datetime.now(timezone.utc)
time_until_expiry = (exp_datetime - now).total_seconds() / 60
# If token expires in less than 5 minutes, try to refresh
if time_until_expiry < TOKEN_REFRESH_THRESHOLD_MINUTES:
if refresh_token:
new_tokens = refresh_access_token(refresh_token)
if new_tokens:
# Update session with new tokens
session['access_token'] = new_tokens['access_token']
session['refresh_token'] = new_tokens['refresh_token']
# Store auth info in request scope
request.scope['auth'] = session.get('user_info')
return None
else:
# Refresh failed, redirect to login
return RedirectResponse('/login', status_code=303)
else:
# No refresh token available, redirect to login
return RedirectResponse('/login', status_code=303)
# Token is valid, store auth info in request scope
request.scope['auth'] = session.get('user_info')
return None
except JWTError:
# Token is invalid or expired, try to refresh
if refresh_token:
new_tokens = refresh_access_token(refresh_token)
if new_tokens:
# Update session with new tokens
session['access_token'] = new_tokens['access_token']
session['refresh_token'] = new_tokens['refresh_token']
request.scope['auth'] = session.get('user_info')
return None
# Could not refresh, redirect to login
return RedirectResponse('/login', status_code=303)
def decode_jwt(token: str) -> Dict[str, Any]:
"""
Decode and validate a JWT token.
Args:
token: JWT token string
Returns:
Dictionary containing the token payload
Raises:
JWTError: If token is invalid or expired
"""
return jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
def check_token_expiry(token: str) -> Optional[float]:
"""
Check how many minutes until token expires.
Args:
token: JWT token string
Returns:
Minutes until expiration, or None if token is invalid
"""
try:
payload = decode_jwt(token)
exp_timestamp = payload.get('exp')
if exp_timestamp:
exp_datetime = datetime.fromtimestamp(exp_timestamp, tz=timezone.utc)
now = datetime.now(timezone.utc)
time_until_expiry = (exp_datetime - now).total_seconds() / 60
return time_until_expiry
return None
except JWTError:
return None
def login_user(email: str, password: str) -> Optional[Dict[str, Any]]:
"""
Authenticate user with email and password.
Args:
email: User email address
password: User password
Returns:
Dictionary containing access_token, refresh_token, and user_info if successful,
None if authentication fails
"""
try:
response = http_client.post(
f"{API_BASE_URL}/auth/login",
data={"username": email, "password": password},
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=10.0
)
if response.status_code == 200:
data = response.json()
return {
'access_token': data.get('access_token'),
'refresh_token': data.get('refresh_token'),
'token_type': data.get('token_type'),
}
return None
except httpx.HTTPError:
return None
def register_user(email: str, username: str, password: str) -> Optional[Dict[str, Any]]:
"""
Register a new user.
Args:
email: User email address
username: User name
password: User password
Returns:
Dictionary containing success message if registration succeeds,
None if registration fails
"""
try:
response = http_client.post(
f"{API_BASE_URL}/auth/register",
json={"email": email, "username": username, "password": password},
timeout=10.0
)
if response.status_code in (200, 201):
return response.json()
return None
except httpx.HTTPError as ex:
return None
def refresh_access_token(refresh_token: str) -> Optional[Dict[str, Any]]:
"""
Refresh the access token using a refresh token.
Args:
refresh_token: Valid refresh token
Returns:
Dictionary containing new access_token and refresh_token if successful,
None if refresh fails
"""
try:
response = http_client.post(
f"{API_BASE_URL}/auth/refresh",
json={"refresh_token": refresh_token},
timeout=10.0
)
if response.status_code == 200:
data = response.json()
return {
'access_token': data.get('access_token'),
'refresh_token': data.get('refresh_token'),
}
return None
except httpx.HTTPError:
return None
def get_user_info(access_token: str) -> Optional[Dict[str, Any]]:
"""
Get current user information using access token.
Args:
access_token: Valid access token
Returns:
Dictionary containing user information if successful,
None if request fails
"""
try:
response = http_client.get(
f"{API_BASE_URL}/auth/me",
headers={"Authorization": f"Bearer {access_token}"},
timeout=10.0
)
if response.status_code == 200:
return response.json()
return None
except httpx.HTTPError:
return None
def logout_user(refresh_token: str) -> bool:
"""
Logout user by revoking the refresh token.
Args:
refresh_token: Refresh token to revoke
Returns:
True if logout successful, False otherwise
"""
try:
response = http_client.post(
f"{API_BASE_URL}/auth/logout",
json={"refresh_token": refresh_token},
timeout=10.0
)
return response.status_code == 200
except httpx.HTTPError:
return False