bellingcat/auto-archiver-api
1
0
Fork 0
mirror of https://github.com/bellingcat/auto-archiver-api.git synced 2026-06-12 05:28:34 +03:00
Code Issues Packages Projects Releases Wiki Activity

added database migrations + soft delete

Browse Source
This commit is contained in:
msramalho
2023-02-27 13:12:22 +01:00
parent 39da4e4eb7
commit 56c18dd96b
16 changed files with 304 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
src/db/crud.py
View File

@@ -15,10 +15,6 @@ def search_tasks_by_url(db: Session, url:str, skip: int = 0, limit: int = 100):
def search_tasks_by_email(db: Session, email:str, skip: int = 0, limit: int = 100):
return base_query(db).filter(models.Task.author==email).offset(skip).limit(limit).all()
def base_query(db:Session):
# allow only some fields to be returned, for example author should remain hidden
return db.query(models.Task).options(load_only(models.Task.id, models.Task.created_at, models.Task.url, models.Task.result))
def create_task(db: Session, task: schemas.TaskCreate):
db_task = models.Task(id=task.id, url=task.url, author=task.author, result=task.result)
db.add(db_task)
@@ -26,10 +22,22 @@ def create_task(db: Session, task: schemas.TaskCreate):
db.refresh(db_task)
return db_task
# TODO: implement soft delete so that S3 content can be found ant not dangling
def delete_task(db: Session, task_id: str, email:str)->bool:
db_task = db.query(models.Task).filter(models.Task.id == task_id, models.Task.author==email).first()
# def delete_task(db: Session, task_id: str, email:str)->bool:
# db_task = db.query(models.Task).filter(models.Task.id == task_id, models.Task.author==email).first()
# if db_task:
# db.delete(db_task)
# db.commit()
# return db_task is not None
def soft_delete_task(db: Session, task_id: str, email:str)->bool:
db_task = db.query(models.Task).filter(models.Task.id == task_id, models.Task.author==email, models.Task.deleted==False).first()
if db_task:
db.delete(db_task)
db_task.deleted = True
db.commit()
return db_task is not None
return db_task is not None
def base_query(db:Session):
# allow only some fields to be returned, for example author should remain hidden
return db.query(models.Task)\
.options(load_only(models.Task.id, models.Task.created_at, models.Task.url, models.Task.result))\
.filter(models.Task.deleted == False)