f63bd5104b
- Created FastAPI application structure with main.py - Implemented SQLAlchemy models for tasks with priority and completion status - Set up database connection with SQLite using absolute paths - Created Alembic migrations for database schema - Added CRUD operations for task management - Implemented health check and root endpoints - Added CORS configuration for cross-origin requests - Created comprehensive API documentation - Added Ruff configuration for code linting - Updated README with installation and usage instructions
34 lines
1.1 KiB
Python
34 lines
1.1 KiB
Python
from sqlalchemy.orm import Session
|
|
from app.models.task import Task as TaskModel
|
|
from app.schemas.task import TaskCreate, TaskUpdate
|
|
|
|
def create_task(db: Session, task: TaskCreate):
|
|
db_task = TaskModel(**task.dict())
|
|
db.add(db_task)
|
|
db.commit()
|
|
db.refresh(db_task)
|
|
return db_task
|
|
|
|
def get_tasks(db: Session, skip: int = 0, limit: int = 100):
|
|
return db.query(TaskModel).offset(skip).limit(limit).all()
|
|
|
|
def get_task(db: Session, task_id: int):
|
|
return db.query(TaskModel).filter(TaskModel.id == task_id).first()
|
|
|
|
def update_task(db: Session, task_id: int, task: TaskUpdate):
|
|
db_task = db.query(TaskModel).filter(TaskModel.id == task_id).first()
|
|
if db_task:
|
|
update_data = task.dict(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(db_task, field, value)
|
|
db.commit()
|
|
db.refresh(db_task)
|
|
return db_task
|
|
|
|
def delete_task(db: Session, task_id: int):
|
|
db_task = db.query(TaskModel).filter(TaskModel.id == task_id).first()
|
|
if db_task:
|
|
db.delete(db_task)
|
|
db.commit()
|
|
return True
|
|
return False |