Implement User and Task Management Endpoints with FastAPI and SQLAlchemy (#1)

* assignment

Signed-off-by: vignesh261201 <[email protected]>

* Docker setup done

Signed-off-by: vignesh261201 <[email protected]>

* Removed the unwanted files

Signed-off-by: vignesh261201 <[email protected]>

---------

Signed-off-by: vignesh261201 <[email protected]>

Author: vignesh261201

Dockerfile

@@ -0,0 +1,20 @@
+# Use an official Python runtime as a parent image
+FROM python:3.11-slim
+
+# Set the working directory
+WORKDIR /app
+
+# Copy the requirements file
+COPY requirements.txt .
+
+# Install dependencies
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy the application code
+COPY . .
+
+# Expose the port FastAPI runs on
+EXPOSE 8000
+
+# Command to run the application
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]

docker-compose.yml

@@ -0,0 +1,16 @@
+services:
+  app:
+    build: .
+    ports:
+      - "8000:8000"
+    volumes:
+      - .:/app
+    command: uvicorn main:app --host 0.0.0.0 --port 8000 --reload
+  db:
+    image: postgres:15.9
+    environment:
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: password
+      POSTGRES_DB: task_management
+    ports:
+      - "5432:5432"

main.py

@@ -0,0 +1,88 @@
+from fastapi import FastAPI, Depends, HTTPException
+from sqlalchemy.orm import Session
+from typing import List, Optional
+from project import models, schemas, database
+
+models.Base.metadata.create_all(bind=database.engine)
+
+app = FastAPI()
+
+
+def get_db():
+    db = database.SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+
+@app.post("/users/", response_model=schemas.User)
+def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
+    db_user = models.User(name=user.name, email=user.email)
+    db.add(db_user)
+    db.commit()
+    db.refresh(db_user)
+    return db_user
+
+
+@app.get("/users/{user_id}/", response_model=schemas.User)
+def get_user(user_id: int, db: Session = Depends(get_db)):
+    user = db.query(models.User).filter(models.User.id == user_id).first()
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found")
+    return user
+
+
+@app.get("/users/", response_model=List[schemas.User])
+def list_users(db: Session = Depends(get_db)):
+    return db.query(models.User).all()
+
+
+@app.post("/tasks/", response_model=schemas.Task)
+def create_task(task: schemas.TaskCreate, db: Session = Depends(get_db)):
+    db_task = models.Task(**task.dict())
+    db.add(db_task)
+    db.commit()
+    db.refresh(db_task)
+    return db_task
+
+
+@app.get("/tasks/{task_id}/", response_model=schemas.Task)
+def get_task(task_id: int, db: Session = Depends(get_db)):
+    task = db.query(models.Task).filter(models.Task.id == task_id).first()
+    if not task:
+        raise HTTPException(status_code=404, detail="Task not found")
+    return task
+
+
+@app.get("/tasks/", response_model=List[schemas.Task])
+def list_tasks(user_id: Optional[int] = None, db: Session = Depends(get_db)):
+    query = db.query(models.Task)
+    if user_id:
+        query = query.filter(models.Task.user_id == user_id)
+    return query.all()
+
+
+@app.put("/tasks/{task_id}/", response_model=schemas.Task)
+def update_task(task_id: int, task: schemas.TaskUpdate, db: Session = Depends(get_db)):
+    db_task = db.query(models.Task).filter(models.Task.id == task_id).first()
+    if not db_task:
+        raise HTTPException(status_code=404, detail="Task not found")
+
+    for key, value in task.dict(exclude_unset=True).items():
+        setattr(db_task, key, value)
+
+    db.commit()
+    db.refresh(db_task)
+    return db_task
+
+
+@app.delete("/tasks/{task_id}/", response_model=schemas.Task)
+def delete_task(task_id: int, db: Session = Depends(get_db)):
+    db_task = db.query(models.Task).filter(models.Task.id == task_id).first()
+    if not db_task:
+        raise HTTPException(status_code=404, detail="Task not found")
+
+    db.delete(db_task)
+    db.commit()
+    return db_task

poetry.lock

@@ -0,0 +1,9 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+SQLALCHEMY_DATABASE_URL = "sqlite:///./database.db"
+
+engine = create_engine(
+    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
+)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

project/__init__.py

@@ -0,0 +1,28 @@
+from sqlalchemy import Column, Integer, String, ForeignKey, DateTime
+from sqlalchemy.orm import relationship
+from sqlalchemy.ext.declarative import declarative_base
+
+Base = declarative_base()
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id = Column(Integer, primary_key=True, index=True)
+    name = Column(String, index=True)
+    email = Column(String, unique=True, index=True)
+
+    tasks = relationship("Task", back_populates="user")
+
+
+class Task(Base):
+    __tablename__ = "tasks"
+
+    id = Column(Integer, primary_key=True, index=True)
+    title = Column(String, index=True)
+    description = Column(String, nullable=True)
+    due_date = Column(DateTime, nullable=True)
+    status = Column(String, default="pending")
+    user_id = Column(Integer, ForeignKey("users.id"))
+
+    user = relationship("User", back_populates="tasks")

project/database.py

@@ -0,0 +1,45 @@
+from pydantic import BaseModel
+from typing import Optional
+from datetime import datetime
+
+
+class UserBase(BaseModel):
+    name: str
+    email: str
+
+
+class UserCreate(UserBase):
+    pass
+
+
+class User(UserBase):
+    id: int
+
+    class Config:
+        from_attributes = True
+
+
+class TaskBase(BaseModel):
+    title: str
+    description: Optional[str] = None
+    due_date: Optional[datetime] = None
+    status: Optional[str] = "pending"
+    user_id: int
+
+
+class TaskCreate(TaskBase):
+    pass
+
+
+class TaskUpdate(BaseModel):
+    title: Optional[str] = None
+    description: Optional[str] = None
+    due_date: Optional[datetime] = None
+    status: Optional[str] = None
+
+
+class Task(TaskBase):
+    id: int
+
+    class Config:
+        from_attributes = True

project/models.py

@@ -0,0 +1,25 @@
+[tool.poetry]
+name = "task management"
+version = "0.1.0"
+description = ""
+authors = ["Vigneshwaran Selvaraj <[email protected]>"]
+readme = "README.md"
+
+[tool.poetry.dependencies]
+python = "^3.9"
+fastapi = "^0.115.5"
+sqlalchemy = "^2.0.36"
+uvicorn = "^0.32.0"
+pydantic = "^2.9.2"
+pytest = "^8.3.3"
+httpx = "^0.27.2"
+
+
+[tool.poetry.group.dev.dependencies]
+flake8 = "^7.1.1"
+black = "^24.10.0"
+isort = "^5.13.2"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

project/schemas.py

@@ -0,0 +1,6 @@
+pywin32==306; platform_system == "Windows"
+pydantic==2.8.2
+uvicorn==0.23.1
+fastapi==0.103.1
+python-dateutil==2.9.0
+sqlalchemy==2.0.36