| name | better-auth-skill |
| description | Configure Better Auth with JWT for secure authentication |
| when-to-use | User signup, login, session management |
Better Auth Skill
Instructions
This skill provides guidance for configuring Better Auth with JWT for secure user authentication.
Project Structure
backend/
├── auth/
│ ├── __init__.py
│ ├── service.py # Auth business logic
│ ├── dependencies.py # FastAPI dependencies
│ ├── routes.py # Auth endpoints
│ └── schemas.py # Pydantic schemas
frontend/
└── src/
├── lib/
│ └── auth.ts # Frontend auth utilities
└── hooks/
└── useAuth.ts # Auth state hook
JWT Configuration
from pydantic_settings import BaseSettings
from datetime import timedelta
from typing import Optional
class AuthSettings(BaseSettings):
SECRET_KEY: str
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
REFRESH_TOKEN_EXPIRE_DAYS: int = 7
class Config:
env_file = ".env"
env_file_encoding = "utf-8"
auth_settings = AuthSettings()
Better Auth Setup
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from models.user import User
from database import AsyncSessionLocal
from sqlalchemy import select
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
"""Create JWT access token."""
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(
to_encode,
auth_settings.SECRET_KEY,
algorithm=ALGORITHM
)
return encoded_jwt
def create_refresh_token(data: dict) -> str:
"""Create JWT refresh token."""
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(days=)
to_encode.update({: expire, : })
jwt.encode(to_encode, auth_settings.SECRET_KEY, algorithm=ALGORITHM)
() -> :
pwd_context.verify(plain_password, hashed_password)
() -> :
pwd_context.(password)
() -> [User]:
AsyncSessionLocal() db:
result = db.execute(
select(User).where(User.email == email)
)
user = result.scalar_one_or_none()
user:
verify_password(password, user.hashed_password):
user
() -> [User]:
AsyncSessionLocal() db:
result = db.execute(
select(User).where(User.email == email)
)
result.scalar_one_or_none()
() -> User:
AsyncSessionLocal() db:
hashed_password = hash_password(password)
user = User(
email=email,
hashed_password=hashed_password,
name=name
)
db.add(user)
db.commit()
db.refresh(user)
user
Auth Dependencies
from datetime import datetime
from typing import Optional
from jose import JWTError, jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from database import get_db
from models.user import User
from auth.service import oauth2_scheme, ALGORITHM, auth_settings
class TokenData:
"""Data extracted from JWT token."""
email: Optional[str] = None
user_id: Optional[int] = None
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db)
) -> User:
"""Get current authenticated user from JWT token."""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(
token,
auth_settings.SECRET_KEY,
algorithms=[ALGORITHM]
)
email: str = payload.get("sub")
if email :
credentials_exception
token_data = TokenData(email=email, user_id=payload.get())
JWTError:
credentials_exception
AsyncSessionLocal() session:
result = session.execute(
select(User).where(User.email == token_data.email)
)
user = result.scalar_one_or_none()
user :
credentials_exception
user.is_active:
HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=
)
user
() -> User:
current_user
() -> TokenData:
:
payload = jwt.decode(
token,
auth_settings.SECRET_KEY,
algorithms=[ALGORITHM]
)
TokenData(
email=payload.get(),
user_id=payload.get()
)
JWTError:
TokenData()
Auth Schemas (Pydantic)
from pydantic import BaseModel, EmailStr
from datetime import datetime
from typing import Optional
class UserCreate(BaseModel):
"""Schema for user registration."""
email: EmailStr
password: str = Field(..., min_length=8)
name: str = Field(..., min_length=1, max_length=100)
class UserLogin(BaseModel):
"""Schema for user login."""
email: EmailStr
password: str
class TokenRefresh(BaseModel):
"""Schema for token refresh."""
refresh_token: str
class Token(BaseModel):
"""Schema for authentication tokens."""
access_token: str
refresh_token: str
token_type: str = "bearer"
class TokenPayload(BaseModel):
"""Schema for token payload."""
sub: str
user_id: int
exp: datetime
class UserResponse(BaseModel):
:
email:
name:
is_active:
created_at: datetime
:
from_attributes =
():
user: UserResponse
tokens: Token
Auth Routes
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import AsyncSession
from database import get_db
from auth.service import (
authenticate_user,
create_user,
create_access_token,
create_refresh_token,
get_user_by_email,
)
from auth.schemas import (
UserCreate,
UserLogin,
Token,
UserResponse,
AuthResponse,
TokenRefresh,
)
from auth.dependencies import get_current_user
from models.user import User
router = APIRouter(prefix="/auth", tags=["Authentication"])
@router.post("/register", response_model=AuthResponse, status_code=status.HTTP_201_CREATED)
async def register(
user_data: UserCreate,
db: AsyncSession = Depends(get_db)
):
"""Register a new user."""
existing_user = await get_user_by_email(user_data.email)
if existing_user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email already registered"
)
user = await create_user(
email=user_data.email,
password=user_data.password,
name=user_data.name
)
access_token = create_access_token(
data={"sub": user.email, "user_id": user.}
)
refresh_token = create_refresh_token(
data={: user.email, : user.}
)
AuthResponse(
user=UserResponse.model_validate(user),
tokens=Token(access_token=access_token, refresh_token=refresh_token)
)
():
user = authenticate_user(form_data.username, form_data.password)
user:
HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=,
headers={: },
)
access_token = create_access_token(
data={: user.email, : user.}
)
refresh_token = create_refresh_token(
data={: user.email, : user.}
)
AuthResponse(
user=UserResponse.model_validate(user),
tokens=Token(access_token=access_token, refresh_token=refresh_token)
)
():
user = authenticate_user(credentials.email, credentials.password)
user:
HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=
)
access_token = create_access_token(
data={: user.email, : user.}
)
refresh_token = create_refresh_token(
data={: user.email, : user.}
)
AuthResponse(
user=UserResponse.model_validate(user),
tokens=Token(access_token=access_token, refresh_token=refresh_token)
)
():
auth.service decode_token, get_user_by_email
token_data = decode_token(refresh_data.refresh_token)
token_data.email:
HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=
)
user = get_user_by_email(token_data.email)
user user.is_active:
HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=
)
access_token = create_access_token(
data={: user.email, : user.}
)
Token(access_token=access_token, refresh_token=refresh_data.refresh_token)
():
UserResponse.model_validate(current_user)
():
{: }
Examples
Frontend Auth Hook
'use client'
import { createContext, useContext, useState, useEffect, ReactNode } from 'react'
interface User {
id: number
email: string
name: string
is_active: boolean
}
interface AuthContextType {
user: User | null
isLoading: boolean
login: (email: string, password: string) => Promise<void>
register: (email: string, password: string, name: string) => Promise<void>
logout: () => void
refreshToken: () => Promise<void>
}
const AuthContext = createContext<AuthContextType | null>(null)
() {
[user, setUser] = useState< | >()
[isLoading, setIsLoading] = ()
( {
token = .()
(token) {
()
} {
()
}
}, [])
= () => {
{
res = (, {
: {
:
}
})
(res.) {
userData = res.()
(userData)
} {
.()
.()
}
} (error) {
.(, error)
} {
()
}
}
= () => {
res = (, {
: ,
: { : },
: .({ email, password })
})
(!res.) {
error = res.()
(error. || )
}
data = res.()
.(, data..)
.(, data..)
(data.)
}
= () => {
res = (, {
: ,
: { : },
: .({ email, password, name })
})
(!res.) {
error = res.()
(error. || )
}
data = res.()
.(, data..)
.(, data..)
(data.)
}
= () => {
.()
.()
()
}
= () => {
refresh_token = .()
(!refresh_token) {
()
}
res = (, {
: ,
: { : },
: .({ refresh_token })
})
(!res.) {
()
}
data = res.()
.(, data.)
}
(
)
}
() {
context = ()
(!context) {
()
}
context
}
Protected Route
'use client'
import { useAuth } from '@/hooks/useAuth'
import { useRouter } from 'next/navigation'
import { useEffect } from 'react'
export function ProtectedRoute({ children }: { children: ReactNode }) {
const { user, isLoading } = useAuth()
const router = useRouter()
useEffect(() => {
if (!isLoading && !user) {
router.push('/auth/login')
}
}, [user, isLoading, router])
if (isLoading) {
return <div>Loading...</div>
}
if (!user) {
return null
}
return children
}
API Client with Auth
const API_BASE = process.env.NEXT_PUBLIC_API_URL || 'http://localhost:8000'
class ApiClient {
private getToken(): string | null {
return localStorage.getItem('access_token')
}
private async request<T>(
endpoint: string,
options: RequestInit = {}
): Promise<T> {
const token = this.getToken()
const headers: HeadersInit = {
'Content-Type': 'application/json',
...(token ? { 'Authorization': `Bearer ${token}` } : {}),
...options.headers,
}
const response = await fetch(`${API_BASE}${endpoint}`, {
...options,
headers,
})
if (response.status === 401) {
const refreshed = await this.refresh()
(refreshed) {
.(endpoint, options)
}
.. =
()
}
(!response.) {
error = response.().( ({}))
(error. || )
}
response.()
}
(): <> {
{
refresh_token = .()
(!refresh_token)
res = (, {
: ,
: { : },
: .({ refresh_token })
})
(!res.)
data = res.()
.(, data.)
} {
}
}
get<T>(: ): <T> {
.<T>(endpoint, { : })
}
post<T>(: , : ): <T> {
.<T>(endpoint, {
: ,
: .(data)
})
}
put<T>(: , : ): <T> {
.<T>(endpoint, {
: ,
: .(data)
})
}
<T>(: ): <T> {
.<T>(endpoint, { : })
}
}
api = ()