85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
from sqlalchemy.orm import Session
|
|
import models
|
|
import schemas
|
|
|
|
# Tasks
|
|
def get_task(db: Session, task_id: int, telegram_user_id: str = None):
|
|
query = db.query(models.Task).filter(models.Task.id == task_id)
|
|
if telegram_user_id:
|
|
query = query.filter(models.Task.telegram_user_id == telegram_user_id)
|
|
return query.first()
|
|
|
|
def get_tasks(db: Session, skip: int = 0, limit: int = 100, telegram_user_id: str = None, status: models.TaskStatus = None):
|
|
query = db.query(models.Task)
|
|
if telegram_user_id:
|
|
query = query.filter(models.Task.telegram_user_id == telegram_user_id)
|
|
if status:
|
|
query = query.filter(models.Task.status == status)
|
|
return query.offset(skip).limit(limit).all()
|
|
|
|
def create_task(db: Session, task: schemas.TaskCreate):
|
|
db_task = models.Task(**task.model_dump())
|
|
db.add(db_task)
|
|
db.commit()
|
|
db.refresh(db_task)
|
|
return db_task
|
|
|
|
def update_task(db: Session, task_id: int, task_update: schemas.TaskUpdate, telegram_user_id: str = None):
|
|
db_task = get_task(db, task_id, telegram_user_id)
|
|
if not db_task:
|
|
return None
|
|
|
|
update_data = task_update.model_dump(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(db_task, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_task)
|
|
return db_task
|
|
|
|
def delete_task(db: Session, task_id: int, telegram_user_id: str = None):
|
|
db_task = get_task(db, task_id, telegram_user_id)
|
|
if db_task:
|
|
db.delete(db_task)
|
|
db.commit()
|
|
return True
|
|
return False
|
|
|
|
# Reminders
|
|
def get_reminder(db: Session, reminder_id: int):
|
|
return db.query(models.Reminder).filter(models.Reminder.id == reminder_id).first()
|
|
|
|
def get_reminders(db: Session, task_id: int = None, skip: int = 0, limit: int = 100):
|
|
query = db.query(models.Reminder)
|
|
if task_id:
|
|
query = query.filter(models.Reminder.task_id == task_id)
|
|
return query.offset(skip).limit(limit).all()
|
|
|
|
def create_reminder(db: Session, reminder: schemas.ReminderCreate):
|
|
db_reminder = models.Reminder(**reminder.model_dump())
|
|
db.add(db_reminder)
|
|
db.commit()
|
|
db.refresh(db_reminder)
|
|
return db_reminder
|
|
|
|
def update_reminder(db: Session, reminder_id: int, reminder_update: schemas.ReminderUpdate):
|
|
db_reminder = get_reminder(db, reminder_id)
|
|
if not db_reminder:
|
|
return None
|
|
|
|
update_data = reminder_update.model_dump(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(db_reminder, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_reminder)
|
|
return db_reminder
|
|
|
|
def delete_reminder(db: Session, reminder_id: int):
|
|
db_reminder = get_reminder(db, reminder_id)
|
|
if db_reminder:
|
|
db.delete(db_reminder)
|
|
db.commit()
|
|
return True
|
|
return False
|