53566813bd
- Add SQLite database configuration - Create Todo model, schemas, and CRUD operations - Implement Todo API endpoints - Add Alembic migration for todo table - Set up database initialization in main.py - Update README with project details and instructions - Add pyproject.toml with Ruff configuration
145 lines
3.5 KiB
Python
145 lines
3.5 KiB
Python
from typing import Any, Dict, Optional, Union
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.todo import Todo
|
|
from app.schemas.todo import TodoCreate, TodoUpdate
|
|
|
|
|
|
def get(db: Session, todo_id: int) -> Optional[Todo]:
|
|
"""
|
|
Get a single todo item by ID.
|
|
|
|
Args:
|
|
db: Database session
|
|
todo_id: ID of the todo item to retrieve
|
|
|
|
Returns:
|
|
Todo object if found, None otherwise
|
|
"""
|
|
return db.query(Todo).filter(Todo.id == todo_id).first()
|
|
|
|
|
|
def get_multi(
|
|
db: Session,
|
|
*,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
filters: Optional[Dict[str, Any]] = None
|
|
) -> list[Todo]:
|
|
"""
|
|
Get multiple todo items with optional filtering, pagination.
|
|
|
|
Args:
|
|
db: Database session
|
|
skip: Number of records to skip (for pagination)
|
|
limit: Maximum number of records to return
|
|
filters: Optional dictionary of filter conditions
|
|
|
|
Returns:
|
|
List of Todo objects
|
|
"""
|
|
query = db.query(Todo)
|
|
|
|
# Apply filters if provided
|
|
if filters:
|
|
if "completed" in filters:
|
|
query = query.filter(Todo.completed == filters["completed"])
|
|
|
|
if "title" in filters:
|
|
query = query.filter(Todo.title.ilike(f"%{filters['title']}%"))
|
|
|
|
return query.order_by(Todo.created_at.desc()).offset(skip).limit(limit).all()
|
|
|
|
|
|
def create(db: Session, *, obj_in: TodoCreate) -> Todo:
|
|
"""
|
|
Create a new todo item.
|
|
|
|
Args:
|
|
db: Database session
|
|
obj_in: TodoCreate schema with the data for the new todo
|
|
|
|
Returns:
|
|
Created Todo object
|
|
"""
|
|
db_obj = Todo(
|
|
title=obj_in.title,
|
|
description=obj_in.description,
|
|
completed=obj_in.completed,
|
|
)
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
|
|
def update(
|
|
db: Session, *, db_obj: Todo, obj_in: Union[TodoUpdate, Dict[str, Any]]
|
|
) -> Todo:
|
|
"""
|
|
Update a todo item.
|
|
|
|
Args:
|
|
db: Database session
|
|
db_obj: Existing Todo object to update
|
|
obj_in: TodoUpdate schema or dict with fields to update
|
|
|
|
Returns:
|
|
Updated Todo object
|
|
"""
|
|
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 field in update_data:
|
|
setattr(db_obj, field, update_data[field])
|
|
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
|
|
def remove(db: Session, *, todo_id: int) -> Optional[Todo]:
|
|
"""
|
|
Delete a todo item.
|
|
|
|
Args:
|
|
db: Database session
|
|
todo_id: ID of the todo item to delete
|
|
|
|
Returns:
|
|
Deleted Todo object if found, None otherwise
|
|
"""
|
|
obj = db.query(Todo).get(todo_id)
|
|
if obj:
|
|
db.delete(obj)
|
|
db.commit()
|
|
return obj
|
|
|
|
|
|
def count(db: Session, *, filters: Optional[Dict[str, Any]] = None) -> int:
|
|
"""
|
|
Count todo items with optional filtering.
|
|
|
|
Args:
|
|
db: Database session
|
|
filters: Optional dictionary of filter conditions
|
|
|
|
Returns:
|
|
Count of Todo objects matching the filters
|
|
"""
|
|
query = db.query(Todo)
|
|
|
|
# Apply filters if provided
|
|
if filters:
|
|
if "completed" in filters:
|
|
query = query.filter(Todo.completed == filters["completed"])
|
|
|
|
if "title" in filters:
|
|
query = query.filter(Todo.title.ilike(f"%{filters['title']}%"))
|
|
|
|
return query.count() |