Add Task Management System with FastAPI and SQLite

Features: - User authentication with JWT tokens - Task and project CRUD operations - Task filtering and organization generated with BackendIM... (backend.im)
2025-05-12 10:14:26 +00:00 · 2025-05-12 10:14:26 +00:00 · e7225e6443
commit e7225e6443
parent 2a0235722f
28 changed files with 1123 additions and 2 deletions
--- a/README.md
+++ b/README.md
@ -1,3 +1,154 @@
-# FastAPI Application
+# Task Management System
-This is a FastAPI application bootstrapped by BackendIM, the AI-powered backend generation platform.
+A RESTful API for managing tasks and projects, built with FastAPI and SQLite.
 ## Features
 - User authentication with JWT tokens
 - CRUD operations for tasks and projects
 - Task prioritization and status tracking
 - Project organization for tasks
 - Filter tasks by status, priority, and project
 ## API Endpoints
 - `GET /health` - Health check endpoint
 - `POST /token` - Login and get access token
 - `POST /users/` - Register a new user
 - `GET /users/me` - Get current user info
 - `GET /tasks/` - List all tasks (with filters)
 - `POST /tasks/` - Create a new task
 - `GET /tasks/{task_id}` - Get a specific task
 - `PUT /tasks/{task_id}` - Update a task
 - `DELETE /tasks/{task_id}` - Delete a task
 - `GET /projects/` - List all projects
 - `POST /projects/` - Create a new project
 - `GET /projects/{project_id}` - Get a specific project
 - `PUT /projects/{project_id}` - Update a project
 - `DELETE /projects/{project_id}` - Delete a project
 ## Getting Started
 ### Prerequisites
 - Python 3.8 or higher
 ### Installation
 1. Clone the repository:
   ```
   git clone https://github.com/yourusername/task-management-system.git
   cd task-management-system
   ```
 2. Install dependencies:
   ```
   pip install -r requirements.txt
   ```
 3. Run database migrations:
   ```
   alembic upgrade head
   ```
 4. Start the application:
   ```
   uvicorn main:app --reload
   ```
 5. Access the API documentation at `http://localhost:8000/docs`
 ## Usage
 ### Authentication
 1. Create a new user:
   ```
   curl -X 'POST' \
     'http://localhost:8000/users/' \
     -H 'Content-Type: application/json' \
     -d '{
       "username": "testuser",
       "email": "user@example.com",
       "password": "password123"
     }'
   ```
 2. Get an access token:
   ```
   curl -X 'POST' \
     'http://localhost:8000/token' \
     -H 'Content-Type: application/x-www-form-urlencoded' \
     -d 'username=testuser&password=password123'
   ```
 3. Use the token for authenticated requests:
   ```
   curl -X 'GET' \
     'http://localhost:8000/users/me' \
     -H 'Authorization: Bearer YOUR_ACCESS_TOKEN'
   ```
 ### Task Management
 1. Create a new task:
   ```
   curl -X 'POST' \
     'http://localhost:8000/tasks/' \
     -H 'Authorization: Bearer YOUR_ACCESS_TOKEN' \
     -H 'Content-Type: application/json' \
     -d '{
       "title": "Task Title",
       "description": "Task Description",
       "priority": "HIGH",
       "status": "TODO"
     }'
   ```
 2. Get all tasks:
   ```
   curl -X 'GET' \
     'http://localhost:8000/tasks/' \
     -H 'Authorization: Bearer YOUR_ACCESS_TOKEN'
   ```
 3. Filter tasks by status:
   ```
   curl -X 'GET' \
     'http://localhost:8000/tasks/?status=TODO' \
     -H 'Authorization: Bearer YOUR_ACCESS_TOKEN'
   ```
 ### Project Management
 1. Create a new project:
   ```
   curl -X 'POST' \
     'http://localhost:8000/projects/' \
     -H 'Authorization: Bearer YOUR_ACCESS_TOKEN' \
     -H 'Content-Type: application/json' \
     -d '{
       "name": "Project Name",
       "description": "Project Description"
     }'
   ```
 2. Create a task in a project:
   ```
   curl -X 'POST' \
     'http://localhost:8000/tasks/' \
     -H 'Authorization: Bearer YOUR_ACCESS_TOKEN' \
     -H 'Content-Type: application/json' \
     -d '{
       "title": "Task Title",
       "description": "Task Description",
       "priority": "HIGH",
       "status": "TODO",
       "project_id": 1
     }'
   ```
 ## API Documentation
 - Swagger UI: `http://localhost:8000/docs`
 - ReDoc: `http://localhost:8000/redoc`
--- a/alembic.ini
+++ b/alembic.ini
@ -0,0 +1,105 @@
 # A generic, single database configuration.
 [alembic]
 # path to migration scripts
 script_location = alembic
 # template used to generate migration file names; The default value is %%(rev)s_%%(slug)s
 # Uncomment the line below if you want the files to be prepended with date and time
 # see https://alembic.sqlalchemy.org/en/latest/tutorial.html#editing-the-ini-file
 # for all available tokens
 # file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s
 # sys.path path, will be prepended to sys.path if present.
 # defaults to the current working directory.
 prepend_sys_path = .
 # timezone to use when rendering the date within the migration file
 # as well as the filename.
 # If specified, requires the python-dateutil library that can be
 # installed by adding `alembic[tz]` to the pip requirements
 # string value is passed to dateutil.tz.gettz()
 # leave blank for localtime
 # timezone =
 # max length of characters to apply to the
 # "slug" field
 # truncate_slug_length = 40
 # set to 'true' to run the environment during
 # the 'revision' command, regardless of autogenerate
 # revision_environment = false
 # set to 'true' to allow .pyc and .pyo files without
 # a source .py file to be detected as revisions in the
 # versions/ directory
 # sourceless = false
 # version location specification; This defaults
 # to alembic/versions.  When using multiple version
 # directories, initial revisions must be specified with --version-path.
 # The path separator used here should be the separator specified by "version_path_separator" below.
 # version_locations = %(here)s/bar:%(here)s/bat:alembic/versions
 # version path separator; As mentioned above, this is the character used to split
 # version_locations. The default within new alembic.ini files is "os", which uses os.pathsep.
 # If this key is omitted entirely, it falls back to the legacy behavior of splitting on spaces and/or commas.
 # Valid values for version_path_separator are:
 #
 # version_path_separator = :
 # version_path_separator = ;
 # version_path_separator = space
 version_path_separator = os  # Use os.pathsep. Default configuration used for new projects.
 # the output encoding used when revision files
 # are written from script.py.mako
 # output_encoding = utf-8
 sqlalchemy.url = sqlite:////app/storage/db/db.sqlite
 [post_write_hooks]
 # post_write_hooks defines scripts or Python functions that are run
 # on newly generated revision scripts.  See the documentation for further
 # detail and examples
 # format using "black" - use the console_scripts runner, against the "black" entrypoint
 # hooks = black
 # black.type = console_scripts
 # black.entrypoint = black
 # black.options = -l 79 REVISION_SCRIPT_FILENAME
 # Logging configuration
 [loggers]
 keys = root,sqlalchemy,alembic
 [handlers]
 keys = console
 [formatters]
 keys = generic
 [logger_root]
 level = WARN
 handlers = console
 qualname =
 [logger_sqlalchemy]
 level = WARN
 handlers =
 qualname = sqlalchemy.engine
 [logger_alembic]
 level = INFO
 handlers =
 qualname = alembic
 [handler_console]
 class = StreamHandler
 args = (sys.stderr,)
 level = NOTSET
 formatter = generic
 [formatter_generic]
 format = %(levelname)-5.5s [%(name)s] %(message)s
 datefmt = %H:%M:%S
--- a/alembic/README
+++ b/alembic/README
@ -0,0 +1 @@
 Generic single-database configuration with an async dbapi.
--- a/alembic/env.py
+++ b/alembic/env.py
@ -0,0 +1,85 @@
 import os
 import sys
 from logging.config import fileConfig
 from sqlalchemy import engine_from_config
 from sqlalchemy import pool
 from alembic import context
 # Add the parent directory to the path so that we can import the app
 sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
 # Import the Base for automatic metadata detection
 from app.database import Base
 from app.models import user, task, project  # Import your models here
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.
 config = context.config
 # Interpret the config file for Python logging.
 # This line sets up loggers basically.
 if config.config_file_name is not None:
    fileConfig(config.config_file_name)
 # add your model's MetaData object here
 # for 'autogenerate' support
 target_metadata = Base.metadata
 # other values from the config, defined by the needs of env.py,
 # can be acquired:
 # my_important_option = config.get_main_option("my_important_option")
 # ... etc.
 def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode.
    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.
    Calls to context.execute() here emit the given string to the
    script output.
    """
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()
 def run_migrations_online() -> None:
    """Run migrations in 'online' mode.
    In this scenario we need to create an Engine
    and associate a connection with the context.
    """
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(
            connection=connection, target_metadata=target_metadata
        )
        with context.begin_transaction():
            context.run_migrations()
 if context.is_offline_mode():
    run_migrations_offline()
 else:
    run_migrations_online()
--- a/alembic/script.py.mako
+++ b/alembic/script.py.mako
@ -0,0 +1,24 @@
 """${message}
 Revision ID: ${up_revision}
 Revises: ${down_revision | comma,n}
 Create Date: ${create_date}
 """
 from alembic import op
 import sqlalchemy as sa
 ${imports if imports else ""}
 # revision identifiers, used by Alembic.
 revision = ${repr(up_revision)}
 down_revision = ${repr(down_revision)}
 branch_labels = ${repr(branch_labels)}
 depends_on = ${repr(depends_on)}
 def upgrade() -> None:
    ${upgrades if upgrades else "pass"}
 def downgrade() -> None:
    ${downgrades if downgrades else "pass"}
--- a/alembic/versions/001_initial_migration.py
+++ b/alembic/versions/001_initial_migration.py
@ -0,0 +1,84 @@
 """Initial migration
 Revision ID: 001_initial_migration
 Revises: 
 Create Date: 2023-05-12 00:00:00.000000
 """
 from alembic import op
 import sqlalchemy as sa
 # revision identifiers, used by Alembic.
 revision = '001_initial_migration'
 down_revision = None
 branch_labels = None
 depends_on = None
 def upgrade() -> None:
    # Create users table
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('username', sa.String(), nullable=False),
        sa.Column('email', sa.String(), nullable=False),
        sa.Column('hashed_password', sa.String(), nullable=False),
        sa.Column('is_active', sa.Boolean(), nullable=False, default=True),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('username'),
        sa.UniqueConstraint('email')
    )
    op.create_index(op.f('ix_users_id'), 'users', ['id'], unique=False)
    op.create_index(op.f('ix_users_username'), 'users', ['username'], unique=True)
    op.create_index(op.f('ix_users_email'), 'users', ['email'], unique=True)
    # Create projects table
    op.create_table(
        'projects',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('name', sa.String(), nullable=False),
        sa.Column('description', sa.Text(), nullable=True),
        sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
        sa.Column('updated_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
        sa.Column('owner_id', sa.Integer(), nullable=False),
        sa.ForeignKeyConstraint(['owner_id'], ['users.id'], ),
        sa.PrimaryKeyConstraint('id')
    )
    op.create_index(op.f('ix_projects_id'), 'projects', ['id'], unique=False)
    op.create_index(op.f('ix_projects_name'), 'projects', ['name'], unique=False)
    # Create tasks table
    op.create_table(
        'tasks',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('title', sa.String(), nullable=False),
        sa.Column('description', sa.Text(), nullable=True),
        sa.Column('priority', sa.Enum('LOW', 'MEDIUM', 'HIGH', name='taskpriority'), nullable=False, default='MEDIUM'),
        sa.Column('status', sa.Enum('TODO', 'IN_PROGRESS', 'DONE', name='taskstatus'), nullable=False, default='TODO'),
        sa.Column('due_date', sa.DateTime(), nullable=True),
        sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
        sa.Column('updated_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
        sa.Column('owner_id', sa.Integer(), nullable=False),
        sa.Column('project_id', sa.Integer(), nullable=True),
        sa.ForeignKeyConstraint(['owner_id'], ['users.id'], ),
        sa.ForeignKeyConstraint(['project_id'], ['projects.id'], ),
        sa.PrimaryKeyConstraint('id')
    )
    op.create_index(op.f('ix_tasks_id'), 'tasks', ['id'], unique=False)
    op.create_index(op.f('ix_tasks_title'), 'tasks', ['title'], unique=False)
 def downgrade() -> None:
    op.drop_index(op.f('ix_tasks_title'), table_name='tasks')
    op.drop_index(op.f('ix_tasks_id'), table_name='tasks')
    op.drop_table('tasks')
    op.drop_index(op.f('ix_projects_name'), table_name='projects')
    op.drop_index(op.f('ix_projects_id'), table_name='projects')
    op.drop_table('projects')
    op.drop_index(op.f('ix_users_email'), table_name='users')
    op.drop_index(op.f('ix_users_username'), table_name='users')
    op.drop_index(op.f('ix_users_id'), table_name='users')
    op.drop_table('users')
--- a/app/init.py
+++ b/app/init.py
--- a/app/auth/init.py
+++ b/app/auth/init.py
--- a/app/auth/dependencies.py
+++ b/app/auth/dependencies.py
@ -0,0 +1,32 @@
 from fastapi import Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordBearer
 from sqlalchemy.orm import Session
 from jose import JWTError, jwt
 from app.database import get_db
 from app.models.user import User
 from app.schemas.token import TokenData
 from app.auth.jwt import verify_token
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    token_data = verify_token(token, credentials_exception)
    user = db.query(User).filter(User.username == token_data.username).first()
    if user is None:
        raise credentials_exception
    return user
 def get_current_active_user(current_user: User = Depends(get_current_user)):
    if not current_user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user
--- a/app/auth/jwt.py
+++ b/app/auth/jwt.py
@ -0,0 +1,33 @@
 from datetime import datetime, timedelta
 from jose import JWTError, jwt
 from typing import Optional
 from app.schemas.token import TokenData
 # to generate a secure secret key: openssl rand -hex 32
 SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
 ALGORITHM = "HS256"
 ACCESS_TOKEN_EXPIRE_MINUTES = 30
 def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    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
 def verify_token(token: str, credentials_exception):
    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)
        return token_data
    except JWTError:
        raise credentials_exception
--- a/app/auth/password.py
+++ b/app/auth/password.py
@ -0,0 +1,11 @@
 from passlib.context import CryptContext
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)
 def get_password_hash(password):
    return pwd_context.hash(password)
--- a/app/database.py
+++ b/app/database.py
@ -0,0 +1,26 @@
 from sqlalchemy import create_engine
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import sessionmaker
 from pathlib import Path
 DB_DIR = Path("/app") / "storage" / "db"
 DB_DIR.mkdir(parents=True, exist_ok=True)
 SQLALCHEMY_DATABASE_URL = f"sqlite:///{DB_DIR}/db.sqlite"
 engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False}
 )
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 Base = declarative_base()
 # Dependency
 def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
--- a/app/models/init.py
+++ b/app/models/init.py
--- a/app/models/project.py
+++ b/app/models/project.py
@ -0,0 +1,22 @@
 from sqlalchemy import Column, Integer, String, ForeignKey, DateTime, Text
 from sqlalchemy.orm import relationship
 from sqlalchemy.sql import func
 from app.database import Base
 class Project(Base):
    __tablename__ = "projects"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(Text, nullable=True)
    created_at = Column(DateTime, default=func.now())
    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())
    # Foreign keys
    owner_id = Column(Integer, ForeignKey("users.id"))
    # Relationships
    owner = relationship("User", back_populates="projects")
    tasks = relationship("Task", back_populates="project")
--- a/app/models/task.py
+++ b/app/models/task.py
@ -0,0 +1,39 @@
 from sqlalchemy import Boolean, Column, Integer, String, ForeignKey, DateTime, Text, Enum
 from sqlalchemy.orm import relationship
 from sqlalchemy.sql import func
 import enum
 from app.database import Base
 class TaskPriority(str, enum.Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
 class TaskStatus(str, enum.Enum):
    TODO = "todo"
    IN_PROGRESS = "in_progress"
    DONE = "done"
 class Task(Base):
    __tablename__ = "tasks"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(Text, nullable=True)
    priority = Column(Enum(TaskPriority), default=TaskPriority.MEDIUM)
    status = Column(Enum(TaskStatus), default=TaskStatus.TODO)
    due_date = Column(DateTime, nullable=True)
    created_at = Column(DateTime, default=func.now())
    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())
    # Foreign keys
    owner_id = Column(Integer, ForeignKey("users.id"))
    project_id = Column(Integer, ForeignKey("projects.id"), nullable=True)
    # Relationships
    owner = relationship("User", back_populates="tasks")
    project = relationship("Project", back_populates="tasks")
--- a/app/models/user.py
+++ b/app/models/user.py
@ -0,0 +1,18 @@
 from sqlalchemy import Boolean, Column, Integer, String
 from sqlalchemy.orm import relationship
 from app.database import Base
 class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)
    # Relationships
    tasks = relationship("Task", back_populates="owner")
    projects = relationship("Project", back_populates="owner")
--- a/app/routers/init.py
+++ b/app/routers/init.py
--- a/app/routers/auth.py
+++ b/app/routers/auth.py
@ -0,0 +1,33 @@
 from datetime import timedelta
 from fastapi import APIRouter, Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordRequestForm
 from sqlalchemy.orm import Session
 from app.database import get_db
 from app.models.user import User
 from app.schemas.token import Token
 from app.auth.password import verify_password
 from app.auth.jwt import create_access_token, ACCESS_TOKEN_EXPIRE_MINUTES
 router = APIRouter(tags=["authentication"])
@router.post("/token", response_model=Token)
 async def login_for_access_token(
    form_data: OAuth2PasswordRequestForm = Depends(), 
    db: Session = Depends(get_db)
 ):
    user = db.query(User).filter(User.username == form_data.username).first()
    if not user or not verify_password(form_data.password, user.hashed_password):
        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"}
--- a/app/routers/projects.py
+++ b/app/routers/projects.py
@ -0,0 +1,122 @@
 from fastapi import APIRouter, Depends, HTTPException, status
 from sqlalchemy.orm import Session
 from typing import List
 from app.database import get_db
 from app.models.project import Project
 from app.models.task import Task
 from app.schemas.project import ProjectCreate, Project as ProjectSchema, ProjectUpdate
 from app.auth.dependencies import get_current_active_user
 from app.models.user import User
 router = APIRouter(
    prefix="/projects",
    tags=["projects"],
 )
@router.post("/", response_model=ProjectSchema, status_code=status.HTTP_201_CREATED)
 def create_project(
    project: ProjectCreate,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_active_user)
 ):
    # Create new project
    db_project = Project(
        **project.dict(),
        owner_id=current_user.id
    )
    db.add(db_project)
    db.commit()
    db.refresh(db_project)
    return db_project
@router.get("/", response_model=List[ProjectSchema])
 def read_projects(
    skip: int = 0,
    limit: int = 100,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_active_user)
 ):
    projects = db.query(Project).filter(
        Project.owner_id == current_user.id
    ).offset(skip).limit(limit).all()
    return projects
@router.get("/{project_id}", response_model=ProjectSchema)
 def read_project(
    project_id: int,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_active_user)
 ):
    project = db.query(Project).filter(
        Project.id == project_id,
        Project.owner_id == current_user.id
    ).first()
    if project is None:
        raise HTTPException(status_code=404, detail="Project not found")
    return project
@router.put("/{project_id}", response_model=ProjectSchema)
 def update_project(
    project_id: int,
    project_update: ProjectUpdate,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_active_user)
 ):
    # Get the project
    db_project = db.query(Project).filter(
        Project.id == project_id,
        Project.owner_id == current_user.id
    ).first()
    if db_project is None:
        raise HTTPException(status_code=404, detail="Project not found")
    # Update project with non-None values from update schema
    update_data = project_update.dict(exclude_unset=True)
    for key, value in update_data.items():
        setattr(db_project, key, value)
    db.commit()
    db.refresh(db_project)
    return db_project
@router.delete("/{project_id}", status_code=status.HTTP_204_NO_CONTENT)
 def delete_project(
    project_id: int,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_active_user)
 ):
    # Get the project
    db_project = db.query(Project).filter(
        Project.id == project_id,
        Project.owner_id == current_user.id
    ).first()
    if db_project is None:
        raise HTTPException(status_code=404, detail="Project not found")
    # Check if there are any tasks associated with this project
    tasks_count = db.query(Task).filter(Task.project_id == project_id).count()
    if tasks_count > 0:
        raise HTTPException(
            status_code=400,
            detail=f"Cannot delete project with {tasks_count} associated tasks. Please delete or reassign them first."
        )
    # Delete the project
    db.delete(db_project)
    db.commit()
    return
--- a/app/routers/tasks.py
+++ b/app/routers/tasks.py
@ -0,0 +1,148 @@
 from fastapi import APIRouter, Depends, HTTPException, status
 from sqlalchemy.orm import Session
 from typing import List, Optional
 from app.database import get_db
 from app.models.task import Task
 from app.models.project import Project
 from app.schemas.task import TaskCreate, Task as TaskSchema, TaskUpdate
 from app.auth.dependencies import get_current_active_user
 from app.models.user import User
 router = APIRouter(
    prefix="/tasks",
    tags=["tasks"],
 )
@router.post("/", response_model=TaskSchema, status_code=status.HTTP_201_CREATED)
 def create_task(
    task: TaskCreate,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_active_user)
 ):
    # Check if project exists and belongs to user if project_id is provided
    if task.project_id:
        project = db.query(Project).filter(
            Project.id == task.project_id,
            Project.owner_id == current_user.id
        ).first()
        if not project:
            raise HTTPException(
                status_code=404,
                detail="Project not found or you don't have access to it"
            )
    # Create new task
    db_task = Task(
        **task.dict(),
        owner_id=current_user.id
    )
    db.add(db_task)
    db.commit()
    db.refresh(db_task)
    return db_task
@router.get("/", response_model=List[TaskSchema])
 def read_tasks(
    skip: int = 0,
    limit: int = 100,
    status: Optional[str] = None,
    priority: Optional[str] = None,
    project_id: Optional[int] = None,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_active_user)
 ):
    query = db.query(Task).filter(Task.owner_id == current_user.id)
    # Apply filters if provided
    if status:
        query = query.filter(Task.status == status)
    if priority:
        query = query.filter(Task.priority == priority)
    if project_id:
        query = query.filter(Task.project_id == project_id)
    tasks = query.offset(skip).limit(limit).all()
    return tasks
@router.get("/{task_id}", response_model=TaskSchema)
 def read_task(
    task_id: int,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_active_user)
 ):
    task = db.query(Task).filter(
        Task.id == task_id,
        Task.owner_id == current_user.id
    ).first()
    if task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    return task
@router.put("/{task_id}", response_model=TaskSchema)
 def update_task(
    task_id: int,
    task_update: TaskUpdate,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_active_user)
 ):
    # Get the task
    db_task = db.query(Task).filter(
        Task.id == task_id,
        Task.owner_id == current_user.id
    ).first()
    if db_task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    # Check if project exists and belongs to user if project_id is provided
    if task_update.project_id is not None:
        project = db.query(Project).filter(
            Project.id == task_update.project_id,
            Project.owner_id == current_user.id
        ).first()
        if not project:
            raise HTTPException(
                status_code=404,
                detail="Project not found or you don't have access to it"
            )
    # Update task with non-None values from update schema
    update_data = task_update.dict(exclude_unset=True)
    for key, value in update_data.items():
        setattr(db_task, key, value)
    db.commit()
    db.refresh(db_task)
    return db_task
@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
 def delete_task(
    task_id: int,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_active_user)
 ):
    # Get the task
    db_task = db.query(Task).filter(
        Task.id == task_id,
        Task.owner_id == current_user.id
    ).first()
    if db_task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    # Delete the task
    db.delete(db_task)
    db.commit()
    return
--- a/app/routers/users.py
+++ b/app/routers/users.py
@ -0,0 +1,45 @@
 from fastapi import APIRouter, Depends, HTTPException, status
 from sqlalchemy.orm import Session
 from app.database import get_db
 from app.models.user import User
 from app.schemas.user import UserCreate, User as UserSchema
 from app.auth.password import get_password_hash
 from app.auth.dependencies import get_current_active_user
 router = APIRouter(
    prefix="/users",
    tags=["users"],
 )
@router.post("/", response_model=UserSchema, status_code=status.HTTP_201_CREATED)
 def create_user(user: UserCreate, db: Session = Depends(get_db)):
    # Check if the username already exists
    db_user_by_username = db.query(User).filter(User.username == user.username).first()
    if db_user_by_username:
        raise HTTPException(status_code=400, detail="Username already registered")
    # Check if the email already exists
    db_user_by_email = db.query(User).filter(User.email == user.email).first()
    if db_user_by_email:
        raise HTTPException(status_code=400, detail="Email already registered")
    # Create the new user
    hashed_password = get_password_hash(user.password)
    db_user = User(
        username=user.username,
        email=user.email,
        hashed_password=hashed_password
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user
@router.get("/me", response_model=UserSchema)
 def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user
--- a/app/schemas/init.py
+++ b/app/schemas/init.py
--- a/app/schemas/project.py
+++ b/app/schemas/project.py
@ -0,0 +1,27 @@
 from pydantic import BaseModel
 from typing import Optional, List
 from datetime import datetime
 class ProjectBase(BaseModel):
    name: str
    description: Optional[str] = None
 class ProjectCreate(ProjectBase):
    pass
 class ProjectUpdate(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
 class Project(ProjectBase):
    id: int
    owner_id: int
    created_at: datetime
    updated_at: datetime
    class Config:
        orm_mode = True
--- a/app/schemas/task.py
+++ b/app/schemas/task.py
@ -0,0 +1,36 @@
 from pydantic import BaseModel
 from typing import Optional, List
 from datetime import datetime
 from app.models.task import TaskPriority, TaskStatus
 class TaskBase(BaseModel):
    title: str
    description: Optional[str] = None
    priority: TaskPriority = TaskPriority.MEDIUM
    status: TaskStatus = TaskStatus.TODO
    due_date: Optional[datetime] = None
    project_id: Optional[int] = None
 class TaskCreate(TaskBase):
    pass
 class TaskUpdate(BaseModel):
    title: Optional[str] = None
    description: Optional[str] = None
    priority: Optional[TaskPriority] = None
    status: Optional[TaskStatus] = None
    due_date: Optional[datetime] = None
    project_id: Optional[int] = None
 class Task(TaskBase):
    id: int
    owner_id: int
    created_at: datetime
    updated_at: datetime
    class Config:
        orm_mode = True
--- a/app/schemas/token.py
+++ b/app/schemas/token.py
@ -0,0 +1,11 @@
 from pydantic import BaseModel
 from typing import Optional
 class Token(BaseModel):
    access_token: str
    token_type: str
 class TokenData(BaseModel):
    username: Optional[str] = None
--- a/app/schemas/user.py
+++ b/app/schemas/user.py
@ -0,0 +1,26 @@
 from pydantic import BaseModel, EmailStr
 from typing import Optional, List
 class UserBase(BaseModel):
    username: str
    email: EmailStr
 class UserCreate(UserBase):
    password: str
 class User(UserBase):
    id: int
    is_active: bool
    class Config:
        orm_mode = True
 class UserInDB(User):
    hashed_password: str
    class Config:
        orm_mode = True
--- a/main.py
+++ b/main.py
@ -0,0 +1,33 @@
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from pathlib import Path
 from app.routers import auth, users, tasks, projects
 app = FastAPI(
    title="Task Management System",
    description="A REST API for managing tasks and projects",
    version="0.1.0",
 )
 app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
 )
 # Include routers
 app.include_router(auth.router)
 app.include_router(users.router)
 app.include_router(projects.router)
 app.include_router(tasks.router)
@app.get("/health", tags=["Health"])
 async def health_check():
    return {"status": "healthy"}
 if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)
--- a/requirements.txt
+++ b/requirements.txt
@ -0,0 +1,9 @@
 fastapi==0.104.1
 uvicorn==0.23.2
 sqlalchemy==2.0.23
 pydantic==2.4.2
 alembic==1.12.1
 python-jose==3.3.0
 passlib==1.7.4
 python-multipart==0.0.6
 bcrypt==4.0.1
		`@ -0,0 +1 @@`
							`Generic single-database configuration with an async dbapi.`