Spaces:

Tahasaif3
/

Taskflow-App

Sleeping

from sqlmodel import SQLModel, Field, Relationship

class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    email: str = Field(unique=True, index=True)
    password_hash: str

    tasks: list["Task"] = Relationship(back_populates="owner")

Forbidden (raw SQLAlchemy):

from sqlalchemy import Column, Integer, String  # NO!

User Data Isolation (CRITICAL)

ALWAYS filter by user_id:

from fastapi import Depends, HTTPException
from sqlmodel import select

async def get_user_tasks(
    user_id: int,
    current_user: User = Depends(get_current_user),
    session: Session = Depends(get_session)
):
    # Verify ownership
    if user_id != current_user.id:
        raise HTTPException(status_code=404)  # NOT 403!

    # Filter by user_id
    statement = select(Task).where(Task.user_id == user_id)
    tasks = session.exec(statement).all()
    return tasks

JWT Authentication

Token Validation:

from jose import jwt, JWTError
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer

security = HTTPBearer()

async def get_current_user(
    token: str = Depends(security)
) -> User:
    try:
        payload = jwt.decode(
            token.credentials,
            settings.BETTER_AUTH_SECRET,
            algorithms=[settings.JWT_ALGORITHM]
        )
        user_id: int = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401)
    except JWTError:
        raise HTTPException(status_code=401)

    user = get_user_from_db(user_id)
    if user is None:
        raise HTTPException(status_code=401)
    return user

Password Security

from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

Project Structure

src/
├── main.py              # FastAPI app, CORS, startup
├── config.py            # Environment variables
├── database.py          # SQLModel engine, session
├── models/
│   ├── user.py          # User SQLModel
│   └── task.py          # Task SQLModel
├── schemas/
│   ├── auth.py          # Request/response schemas
│   └── task.py          # Request/response schemas
├── routers/
│   ├── auth.py          # /api/auth/* endpoints
│   └── tasks.py         # /api/{user_id}/tasks/* endpoints
├── middleware/
│   └── auth.py          # JWT validation
└── utils/
    ├── security.py      # bcrypt, JWT helpers
    └── deps.py          # Dependency injection

API Patterns

Endpoint Structure

from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import Session

router = APIRouter(prefix="/api/{user_id}/tasks", tags=["tasks"])

@router.get("/", response_model=list[TaskResponse])
async def list_tasks(
    user_id: int,
    current_user: User = Depends(get_current_user),
    session: Session = Depends(get_session)
):
    # Authorization check
    if user_id != current_user.id:
        raise HTTPException(status_code=404)

    # Query with user_id filter
    statement = select(Task).where(Task.user_id == user_id)
    tasks = session.exec(statement).all()
    return tasks

Error Responses

# 401 Unauthorized - Invalid/missing JWT
raise HTTPException(
    status_code=401,
    detail="Invalid authentication credentials"
)

# 404 Not Found - Resource doesn't exist OR unauthorized access
raise HTTPException(
    status_code=404,
    detail="Task not found"
)

# 400 Bad Request - Validation error
raise HTTPException(
    status_code=400,
    detail="Title must be between 1-200 characters"
)

# 409 Conflict - Duplicate resource
raise HTTPException(
    status_code=409,
    detail="An account with this email already exists"
)

Database Migrations

Creating Migrations:

uv run alembic revision --autogenerate -m "Add users and tasks tables"

Applying Migrations:

uv run alembic upgrade head

Migration File Structure:

def upgrade():
    op.create_table(
        'user',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('email', sa.String(), unique=True),
        sa.Column('password_hash', sa.String()),
    )
    op.create_index('ix_user_email', 'user', ['email'])

Testing

Fixtures (tests/conftest.py):

import pytest
from sqlmodel import Session, create_engine
from fastapi.testclient import TestClient

@pytest.fixture
def session():
    engine = create_engine("sqlite:///:memory:")
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session

@pytest.fixture
def client(session):
    app.dependency_overrides[get_session] = lambda: session
    yield TestClient(app)

Test Example:

def test_create_task(client, auth_headers):
    response = client.post(
        "/api/1/tasks",
        headers=auth_headers,
        json={"title": "Test Task", "description": "Test"}
    )
    assert response.status_code == 201
    assert response.json()["title"] == "Test Task"

Environment Variables

Required in .env:

DATABASE_URL=postgresql://taskuser:taskpassword@db:5432/taskdb
BETTER_AUTH_SECRET=your-secret-key-change-in-production
JWT_SECRET_KEY=your-jwt-secret-change-in-production
JWT_ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_DAYS=7

Common Mistakes to Avoid

❌ Using raw SQLAlchemy instead of SQLModel ✅ Use SQLModel for all database models

❌ Trusting user_id from request parameters ✅ Always extract from validated JWT token

❌ Returning 403 for unauthorized access ✅ Return 404 to prevent information leakage

❌ SQL string concatenation ✅ SQLModel parameterized queries only

❌ Plaintext passwords ✅ bcrypt hashing always

References

Root Instructions: ../CLAUDE.md
Feature Spec: ../specs/001-task-crud-auth/spec.md
Constitution: ../.specify/memory/constitution.md