f8bb3dd21d
- Set up FastAPI project structure with API versioning - Create database models for users and tasks - Implement SQLAlchemy ORM with SQLite database - Initialize Alembic for database migrations - Create API endpoints for task management (CRUD) - Create API endpoints for user management - Add JWT authentication and authorization - Add health check endpoint - Add comprehensive README.md with API documentation
85 lines
2.0 KiB
Python
85 lines
2.0 KiB
Python
from typing import Any, Dict, List, Optional, Union
|
|
|
|
from sqlalchemy import and_
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.task import Task
|
|
from app.schemas.task import TaskCreate, TaskUpdate
|
|
|
|
|
|
def get(db: Session, task_id: int, user_id: int) -> Optional[Task]:
|
|
"""
|
|
Get a task by ID and user_id
|
|
"""
|
|
return db.query(Task).filter(
|
|
and_(Task.id == task_id, Task.user_id == user_id, not Task.is_deleted)
|
|
).first()
|
|
|
|
def get_multi(
|
|
db: Session, user_id: int, skip: int = 0, limit: int = 100
|
|
) -> List[Task]:
|
|
"""
|
|
Get multiple tasks by user_id
|
|
"""
|
|
return db.query(Task).filter(
|
|
and_(Task.user_id == user_id, not Task.is_deleted)
|
|
).offset(skip).limit(limit).all()
|
|
|
|
def create(db: Session, obj_in: TaskCreate, user_id: int) -> Task:
|
|
"""
|
|
Create a new task
|
|
"""
|
|
db_obj = Task(
|
|
**obj_in.model_dump(),
|
|
user_id=user_id
|
|
)
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def update(
|
|
db: Session, db_obj: Task, obj_in: Union[TaskUpdate, Dict[str, Any]]
|
|
) -> Task:
|
|
"""
|
|
Update a task
|
|
"""
|
|
if isinstance(obj_in, dict):
|
|
update_data = obj_in
|
|
else:
|
|
update_data = obj_in.model_dump(exclude_unset=True)
|
|
|
|
for field in update_data:
|
|
if hasattr(db_obj, field):
|
|
setattr(db_obj, field, update_data[field])
|
|
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def delete(db: Session, task_id: int, user_id: int) -> bool:
|
|
"""
|
|
Soft delete a task by setting is_deleted to True
|
|
"""
|
|
task = get(db, task_id=task_id, user_id=user_id)
|
|
if not task:
|
|
return False
|
|
|
|
task.is_deleted = True
|
|
db.add(task)
|
|
db.commit()
|
|
return True
|
|
|
|
def hard_delete(db: Session, task_id: int, user_id: int) -> bool:
|
|
"""
|
|
Hard delete a task
|
|
"""
|
|
task = get(db, task_id=task_id, user_id=user_id)
|
|
if not task:
|
|
return False
|
|
|
|
db.delete(task)
|
|
db.commit()
|
|
return True
|