d60cfb5c4f
- Set up FastAPI project structure - Create Task model with SQLAlchemy - Set up Alembic for migrations - Create CRUD operations for tasks - Implement API endpoints for tasks - Add health check endpoint - Update documentation generated with BackendIM... (backend.im)
54 lines
1.3 KiB
Python
54 lines
1.3 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.task import Task
|
|
from app.schemas.task import TaskCreate, TaskUpdate
|
|
|
|
|
|
def get_task(db: Session, task_id: int) -> Optional[Task]:
|
|
return db.query(Task).filter(Task.id == task_id).first()
|
|
|
|
|
|
def get_tasks(
|
|
db: Session,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
is_completed: Optional[bool] = None
|
|
) -> List[Task]:
|
|
query = db.query(Task)
|
|
|
|
if is_completed is not None:
|
|
query = query.filter(Task.is_completed == is_completed)
|
|
|
|
return query.order_by(Task.created_at.desc()).offset(skip).limit(limit).all()
|
|
|
|
|
|
def create_task(db: Session, task_in: TaskCreate) -> Task:
|
|
db_task = Task(
|
|
title=task_in.title,
|
|
description=task_in.description,
|
|
is_completed=task_in.is_completed,
|
|
priority=task_in.priority,
|
|
due_date=task_in.due_date,
|
|
)
|
|
db.add(db_task)
|
|
db.commit()
|
|
db.refresh(db_task)
|
|
return db_task
|
|
|
|
|
|
def update_task(db: Session, task: Task, task_in: TaskUpdate) -> Task:
|
|
update_data = task_in.dict(exclude_unset=True)
|
|
|
|
for field, value in update_data.items():
|
|
setattr(task, field, value)
|
|
|
|
db.add(task)
|
|
db.commit()
|
|
db.refresh(task)
|
|
return task
|
|
|
|
|
|
def delete_task(db: Session, task: Task) -> None:
|
|
db.delete(task)
|
|
db.commit() |