backend.im/simpletodoapp-x9cyfn
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add user authentication with JWT and update todos to be user-specific

Browse Source
This commit is contained in:
Automated Action 2025-05-16 01:01:54 +00:00
parent fb99c09cdd
commit 1710f43c85
13 changed files with 595 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

69
README.md
View File

@ -1,10 +1,13 @@
# SimpleTodoApp API
A FastAPI-based backend for a simple Todo application with SQLite database.
A FastAPI-based backend for a simple Todo application with SQLite database and user authentication.
## Features
- Create, read, update, and delete Todo items
- User registration and authentication with JWT tokens
- Create, read, update, and delete Todo items (protected by authentication)
- User-specific Todo items
- Role-based access control (regular users and superusers)
- Health check endpoint
- SQLite database with SQLAlchemy ORM
- Database migrations with Alembic
@ -16,8 +19,19 @@ A FastAPI-based backend for a simple Todo application with SQLite database.
simpletodoapp/
├── api/ # API-related code
│ ├── crud/ # CRUD operations
│ │ ├── todo.py # Todo CRUD operations
│ │ └── user.py # User CRUD operations
│ ├── routers/ # API endpoints
│ └── schemas/ # Pydantic models for request/response validation
│ │ ├── auth_router.py # Authentication endpoints
│ │ ├── health_router.py # Health check endpoint
│ │ ├── todo_router.py # Todo endpoints
│ │ └── user_router.py # User endpoints
│ ├── schemas/ # Pydantic models for request/response validation
│ │ ├── health.py # Health check schemas
│ │ ├── todo.py # Todo schemas
│ │ └── user.py # User and authentication schemas
│ └── utils/ # Utility functions
│ └── auth.py # Authentication utilities
├── db/ # Database-related code
│ ├── database.py # Database connection and session
│ └── models.py # SQLAlchemy models
@ -34,11 +48,28 @@ simpletodoapp/
```
pip install -r requirements.txt
```
3. Run the application:
3. Apply database migrations:
```
alembic upgrade head
```
4. Run the application:
```
uvicorn main:app --reload
```
## Authentication
The API uses JWT (JSON Web Tokens) for authentication. To use protected endpoints:
1. Register a new user using `POST /api/users`
2. Get an access token using `POST /api/auth/token` with your username and password
3. Include the token in the `Authorization` header of your requests:
```
Authorization: Bearer <your_token>
```
Access tokens expire after 30 minutes by default.
## API Documentation
Once the server is running, you can access:
@ -47,9 +78,29 @@ Once the server is running, you can access:
## API Endpoints
### Authentication
- `POST /api/auth/token` - Get access token (login)
- `POST /api/users` - Register a new user
### Users
- `GET /api/users/me` - Get current user information
- `PUT /api/users/me` - Update current user information
- `GET /api/users/{id}` - Get user information by ID (current user or superuser only)
- `GET /api/users` - List all users (superuser only)
- `DELETE /api/users/{id}` - Delete a user (superuser only)
### Todos
All todo endpoints require authentication.
- `GET /api/todos` - List all todos for current user
- `GET /api/todos/{id}` - Get a single todo by ID (owned by current user)
- `POST /api/todos` - Create a new todo (owned by current user)
- `PATCH /api/todos/{id}` - Update a todo (owned by current user)
- `DELETE /api/todos/{id}` - Delete a todo (owned by current user)
### Health
- `GET /api/health` - Health check endpoint
- `GET /api/todos` - List all todos
- `GET /api/todos/{id}` - Get a single todo by ID
- `POST /api/todos` - Create a new todo
- `PATCH /api/todos/{id}` - Update a todo (partial update)
- `DELETE /api/todos/{id}` - Delete a todo

30
api/crud/todo.py
View File

@ -6,36 +6,36 @@ from api.schemas.todo import TodoCreate, TodoUpdate
from db.models import Todo
def get_todos(db: Session, skip: int = 0, limit: int = 100) -> list[Todo]:
def get_todos(db: Session, user_id: int, skip: int = 0, limit: int = 100) -> list[Todo]:
"""
Get all todos with pagination
Get all todos for a user with pagination
"""
return db.query(Todo).offset(skip).limit(limit).all()
return db.query(Todo).filter(Todo.owner_id == user_id).offset(skip).limit(limit).all()
def get_todo(db: Session, todo_id: int) -> Optional[Todo]:
def get_todo(db: Session, todo_id: int, user_id: int) -> Optional[Todo]:
"""
Get a specific todo by ID
Get a specific todo by ID for a specific user
"""
return db.query(Todo).filter(Todo.id == todo_id).first()
return db.query(Todo).filter(Todo.id == todo_id, Todo.owner_id == user_id).first()
def create_todo(db: Session, todo: TodoCreate) -> Todo:
def create_todo(db: Session, todo: TodoCreate, user_id: int) -> Todo:
"""
Create a new todo
Create a new todo for a user
"""
db_todo = Todo(**todo.dict())
db_todo = Todo(**todo.dict(), owner_id=user_id)
db.add(db_todo)
db.commit()
db.refresh(db_todo)
return db_todo
def update_todo(db: Session, todo_id: int, todo: TodoUpdate) -> Optional[Todo]:
def update_todo(db: Session, todo_id: int, todo: TodoUpdate, user_id: int) -> Optional[Todo]:
"""
Update an existing todo
Update an existing todo for a user
"""
db_todo = get_todo(db, todo_id)
db_todo = get_todo(db, todo_id, user_id)
if db_todo is None:
return None
@ -49,12 +49,12 @@ def update_todo(db: Session, todo_id: int, todo: TodoUpdate) -> Optional[Todo]:
return db_todo
def delete_todo(db: Session, todo_id: int) -> bool:
def delete_todo(db: Session, todo_id: int, user_id: int) -> bool:
"""
Delete a todo by ID
Delete a todo by ID for a user
Returns True if the todo was deleted, False if it didn't exist
"""
db_todo = get_todo(db, todo_id)
db_todo = get_todo(db, todo_id, user_id)
if db_todo is None:
return False

84
api/crud/user.py Normal file
View File

@ -0,0 +1,84 @@
from typing import Optional
from sqlalchemy.orm import Session
from api.schemas.user import UserCreate, UserUpdate
from api.utils.auth import get_password_hash
from db.models import User
def get_user(db: Session, user_id: int) -> Optional[User]:
"""
Get a user by ID
"""
return db.query(User).filter(User.id == user_id).first()
def get_user_by_email(db: Session, email: str) -> Optional[User]:
"""
Get a user by email
"""
return db.query(User).filter(User.email == email).first()
def get_user_by_username(db: Session, username: str) -> Optional[User]:
"""
Get a user by username
"""
return db.query(User).filter(User.username == username).first()
def get_users(db: Session, skip: int = 0, limit: int = 100) -> list[User]:
"""
Get all users with pagination
"""
return db.query(User).offset(skip).limit(limit).all()
def create_user(db: Session, user: UserCreate) -> User:
"""
Create a new user
"""
hashed_password = get_password_hash(user.password)
db_user = User(
email=user.email,
username=user.username,
hashed_password=hashed_password,
)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
def update_user(db: Session, user_id: int, user: UserUpdate) -> Optional[User]:
"""
Update an existing user
"""
db_user = get_user(db, user_id)
if db_user is None:
return None
update_data = user.dict(exclude_unset=True)
if "password" in update_data:
update_data["hashed_password"] = get_password_hash(update_data.pop("password"))
for key, value in update_data.items():
setattr(db_user, key, value)
db.commit()
db.refresh(db_user)
return db_user
def delete_user(db: Session, user_id: int) -> bool:
"""
Delete a user
"""
db_user = get_user(db, user_id)
if db_user is None:
return False
db.delete(db_user)
db.commit()
return True

39
api/routers/auth_router.py Normal file
View File

@ -0,0 +1,39 @@
from datetime import timedelta
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from api.schemas.user import Token
from api.utils.auth import (
ACCESS_TOKEN_EXPIRE_MINUTES,
authenticate_user,
create_access_token,
)
from db.database import get_db
router = APIRouter()
@router.post("/token", response_model=Token)
def login_for_access_token(
form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db),
) -> Any:
"""
Generate JWT token for authenticated user
"""
user = authenticate_user(db, form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires,
)
return {"access_token": access_token, "token_type": "bearer"}

54
api/routers/todo_router.py
View File

@ -3,56 +3,80 @@ from sqlalchemy.orm import Session
from api.crud.todo import create_todo, delete_todo, get_todo, get_todos, update_todo
from api.schemas.todo import TodoCreate, TodoResponse, TodoUpdate
from api.utils.auth import get_current_active_user
from db.database import get_db
from db.models import User
router = APIRouter()
@router.get("/", response_model=list[TodoResponse])
def read_todos(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
def read_todos(
skip: int = 0,
limit: int = 100,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_active_user),
):
"""
Get all todos with pagination
Get all todos for the current user with pagination
"""
todos = get_todos(db, skip=skip, limit=limit)
todos = get_todos(db, user_id=current_user.id, skip=skip, limit=limit)
return todos
@router.get("/{todo_id}", response_model=TodoResponse)
def read_todo(todo_id: int, db: Session = Depends(get_db)):
def read_todo(
todo_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_active_user),
):
"""
Get a specific todo by ID
Get a specific todo by ID for the current user
"""
db_todo = get_todo(db, todo_id=todo_id)
db_todo = get_todo(db, todo_id=todo_id, user_id=current_user.id)
if db_todo is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found")
return db_todo
@router.post("/", response_model=TodoResponse, status_code=status.HTTP_201_CREATED)
def create_new_todo(todo: TodoCreate, db: Session = Depends(get_db)):
def create_new_todo(
todo: TodoCreate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_active_user),
):
"""
Create a new todo
Create a new todo for the current user
"""
return create_todo(db=db, todo=todo)
return create_todo(db=db, todo=todo, user_id=current_user.id)
@router.patch("/{todo_id}", response_model=TodoResponse)
def update_existing_todo(todo_id: int, todo: TodoUpdate, db: Session = Depends(get_db)):
def update_existing_todo(
todo_id: int,
todo: TodoUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_active_user),
):
"""
Update an existing todo (partial update)
Update an existing todo (partial update) for the current user
"""
db_todo = update_todo(db=db, todo_id=todo_id, todo=todo)
db_todo = update_todo(db=db, todo_id=todo_id, todo=todo, user_id=current_user.id)
if db_todo is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found")
return db_todo
@router.delete("/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_existing_todo(todo_id: int, db: Session = Depends(get_db)):
def delete_existing_todo(
todo_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_active_user),
):
"""
Delete a todo
Delete a todo for the current user
"""
success = delete_todo(db=db, todo_id=todo_id)
success = delete_todo(db=db, todo_id=todo_id, user_id=current_user.id)
if not success:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found")
return None

110
api/routers/user_router.py Normal file
View File

@ -0,0 +1,110 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from api.crud.user import (
create_user,
delete_user,
get_user,
get_user_by_email,
get_user_by_username,
get_users,
update_user,
)
from api.schemas.user import UserCreate, UserResponse, UserUpdate
from api.utils.auth import get_current_active_user, get_current_superuser
from db.database import get_db
from db.models import User
router = APIRouter()
@router.get("/", response_model=list[UserResponse])
def read_users(
skip: int = 0,
limit: int = 100,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_superuser),
) -> list[User]:
"""
Get all users with pagination (superuser only)
"""
users = get_users(db, skip=skip, limit=limit)
return users
@router.get("/me", response_model=UserResponse)
def read_user_me(current_user: User = Depends(get_current_active_user)) -> User:
"""
Get current user
"""
return current_user
@router.put("/me", response_model=UserResponse)
def update_user_me(
user: UserUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_active_user),
) -> User:
"""
Update current user
"""
# Prevent user from changing themselves to superuser
if user.is_superuser is not None:
user.is_superuser = current_user.is_superuser
return update_user(db, current_user.id, user)
@router.get("/{user_id}", response_model=UserResponse)
def read_user(
user_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_active_user),
) -> User:
"""
Get a specific user by ID (superuser or same user)
"""
db_user = get_user(db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
if current_user.id != user_id and not current_user.is_superuser:
raise HTTPException(status_code=403, detail="Not enough permissions")
return db_user
@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
def register_user(user: UserCreate, db: Session = Depends(get_db)) -> User:
"""
Register a new user
"""
db_user_by_email = get_user_by_email(db, email=user.email)
if db_user_by_email:
raise HTTPException(
status_code=400,
detail="Email already registered",
)
db_user_by_username = get_user_by_username(db, username=user.username)
if db_user_by_username:
raise HTTPException(
status_code=400,
detail="Username already taken",
)
return create_user(db=db, user=user)
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user_by_id(
user_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_superuser),
) -> None:
"""
Delete a user (superuser only)
"""
success = delete_user(db=db, user_id=user_id)
if not success:
raise HTTPException(status_code=404, detail="User not found")
return None

51
api/schemas/user.py Normal file
View File

@ -0,0 +1,51 @@
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, EmailStr, Field
class UserBase(BaseModel):
"""Base User schema with common attributes"""
email: EmailStr = Field(..., description="User email address")
username: str = Field(..., min_length=3, max_length=50, description="User username")
class UserCreate(UserBase):
"""Schema for creating a new user"""
password: str = Field(..., min_length=8, description="User password")
class UserUpdate(BaseModel):
"""Schema for updating an existing user, all fields are optional"""
email: Optional[EmailStr] = Field(None, description="User email address")
username: Optional[str] = Field(
None, min_length=3, max_length=50, description="User username",
)
password: Optional[str] = Field(None, min_length=8, description="User password")
is_active: Optional[bool] = Field(None, description="User active status")
is_superuser: Optional[bool] = Field(None, description="User superuser status")
class UserResponse(UserBase):
"""Schema for user response that includes database fields"""
id: int
is_active: bool
is_superuser: bool
created_at: datetime
updated_at: Optional[datetime] = None
class Config:
"""ORM mode config for the UserResponse schema"""
orm_mode = True
from_attributes = True
class Token(BaseModel):
"""Schema for JWT token"""
access_token: str
token_type: str
class TokenData(BaseModel):
"""Schema for JWT token data"""
username: Optional[str] = None

1
api/utils/__init__.py Normal file
View File

@ -0,0 +1 @@
"""API utility functions"""

94
api/utils/auth.py Normal file
View File

@ -0,0 +1,94 @@
from datetime import datetime, timedelta
from typing import Optional, Union
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from passlib.context import CryptContext
from sqlalchemy.orm import Session
from api.schemas.user import TokenData
from db.database import get_db
from db.models import User
# to get a string like this run: openssl rand -hex 32
# In production, this should be stored in environment variables
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7" # noqa: S105
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/auth/token")
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""Verify a password against a hash"""
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
"""Hash a password"""
return pwd_context.hash(password)
def authenticate_user(db: Session, username: str, password: str) -> Union[User, bool]:
"""Authenticate a user"""
user = db.query(User).filter(User.username == username).first()
if not user or not verify_password(password, user.hashed_password):
return False
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, detail="Inactive user",
)
return user
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
"""Create a JWT 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, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
async def get_current_user(
token: str = Depends(oauth2_scheme), db: Session = Depends(get_db),
) -> User:
"""Get the current user from a 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, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
token_data = TokenData(username=username)
except JWTError as err:
raise credentials_exception from err
user = db.query(User).filter(User.username == token_data.username).first()
if user is None or not user.is_active:
raise credentials_exception
return user
async def get_current_active_user(current_user: User = Depends(get_current_user)) -> User:
"""Get the current active user"""
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
async def get_current_superuser(current_user: User = Depends(get_current_user)) -> User:
"""Get the current superuser"""
if not current_user.is_superuser:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions",
)
return current_user

25
db/models.py
View File

@ -1,9 +1,28 @@
from sqlalchemy import Boolean, Column, DateTime, Integer, String
from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from .database import Base
class User(Base):
"""
User model for authentication and authorization
"""
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True, nullable=False)
username = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
is_active = Column(Boolean, default=True)
is_superuser = Column(Boolean, default=False)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
todos = relationship("Todo", back_populates="owner", cascade="all, delete-orphan")
class Todo(Base):
"""
Todo model representing a task to be done
@ -16,3 +35,7 @@ class Todo(Base):
completed = Column(Boolean, default=False)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
# Add relationship to User
owner_id = Column(Integer, ForeignKey("users.id"), nullable=False)
owner = relationship("User", back_populates="todos")

6
main.py
View File

@ -2,12 +2,12 @@ import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from api.routers import health_router, todo_router
from api.routers import auth_router, health_router, todo_router, user_router
from db.database import create_tables
app = FastAPI(
title="SimpleTodoApp API",
description="API for a simple todo application",
description="API for a simple todo application with authentication",
version="0.1.0",
)
@ -21,6 +21,8 @@ app.add_middleware(
)
# Include routers
app.include_router(auth_router.router, prefix="/api/auth", tags=["auth"])
app.include_router(user_router.router, prefix="/api/users", tags=["users"])
app.include_router(todo_router.router, prefix="/api/todos", tags=["todos"])
app.include_router(health_router.router, prefix="/api", tags=["health"])

69
migrations/versions/2_add_users_table_and_update_todos.py Normal file
View File

@ -0,0 +1,69 @@
"""Add users table and update todos
Revision ID: 2_add_users_table_and_update_todos
Revises: 1_initial_create_todos_table
Create Date: 2023-07-20 10:00:00.000000
"""
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision = '2_add_users_table_and_update_todos'
down_revision = '1_initial_create_todos_table'
branch_labels = None
depends_on = None
def upgrade():
# Create users table
op.create_table(
'users',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('email', sa.String(), nullable=False),
sa.Column('username', sa.String(), nullable=False),
sa.Column('hashed_password', sa.String(), nullable=False),
sa.Column('is_active', sa.Boolean(), nullable=True, default=True),
sa.Column('is_superuser', sa.Boolean(), nullable=True, default=False),
sa.Column(
'created_at',
sa.DateTime(timezone=True),
server_default=sa.text('(CURRENT_TIMESTAMP)'),
nullable=True,
),
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_users_email'), 'users', ['email'], unique=True)
op.create_index(op.f('ix_users_id'), 'users', ['id'], unique=False)
op.create_index(op.f('ix_users_username'), 'users', ['username'], unique=True)
# Add owner_id column to todos table
op.add_column('todos', sa.Column('owner_id', sa.Integer(), nullable=True))
# Create foreign key from todos to users
op.create_foreign_key(
'fk_todos_owner_id_users',
'todos', 'users',
['owner_id'], ['id'],
)
# Update all existing todos to have owner_id = 1 (first user)
op.execute("UPDATE todos SET owner_id = 1")
# Make owner_id non-nullable after updating existing todos
op.alter_column('todos', 'owner_id', nullable=False)
def downgrade():
# Drop foreign key constraint
op.drop_constraint('fk_todos_owner_id_users', 'todos', type_='foreignkey')
# Drop owner_id column from todos
op.drop_column('todos', 'owner_id')
# Drop users table
op.drop_index(op.f('ix_users_username'), table_name='users')
op.drop_index(op.f('ix_users_id'), table_name='users')
op.drop_index(op.f('ix_users_email'), table_name='users')
op.drop_table('users')

3
requirements.txt
View File

@ -7,3 +7,6 @@ ruff>=0.0.280
python-dotenv>=1.0.0
python-multipart>=0.0.6
pathlib>=1.0.1
passlib>=1.7.4
python-jose>=3.3.0
bcrypt>=4.0.1