bellingcat/auto-archiver-api
1
0
Fork 0
mirror of https://github.com/bellingcat/auto-archiver-api.git synced 2026-06-11 21:18:35 +03:00
Code Issues Packages Projects Releases Wiki Activity

switching from optional response_model to mandatory return type

Browse Source
This commit is contained in:
msramalho
2024-10-29 16:17:40 +00:00
parent aaada7d83f
commit 8c658cdf52
7 changed files with 35 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
src/endpoints/url.py
View File

@@ -15,8 +15,8 @@ from worker.main import create_archive_task
url_router = APIRouter(prefix="/url", tags=["Single URL operations"])
@url_router.post("/archive", status_code=201, summary="Submit a single URL archive request, starts an archiving task.", response_model=schemas.Task, response_description="task_id for the archiving task, will match the archive id.")
def archive_url(archive: schemas.ArchiveCreate, email=Depends(get_token_or_user_auth)):
@url_router.post("/archive", status_code=201, summary="Submit a single URL archive request, starts an archiving task.", response_description="task_id for the archiving task, will match the archive id.")
def archive_url(archive: schemas.ArchiveCreate, email=Depends(get_token_or_user_auth)) -> schemas.Task:
archive.author_id = email
url = archive.url
logger.info(f"new {archive.public=} task for {email=} and {archive.group_id=}: {url}")
@@ -28,30 +28,31 @@ def archive_url(archive: schemas.ArchiveCreate, email=Depends(get_token_or_user_
return JSONResponse(task_response.model_dump(), status_code=201)
@url_router.get("/search", response_model=list[schemas.Archive], summary="Search for archive entries by URL.")
@url_router.get("/search", summary="Search for archive entries by URL.")
def search_by_url(
url: str, skip: int = 0, limit: int = 25,
archived_after: datetime = None, archived_before: datetime = None,
db: Session = Depends(get_db_dependency),
email=Depends(get_token_or_user_auth)):
email=Depends(get_token_or_user_auth)
) -> list[schemas.ArchiveResult]:
return crud.search_archives_by_url(db, url.strip(), email, skip=skip, limit=limit, archived_after=archived_after, archived_before=archived_before)
@url_router.get("/latest", response_model=list[schemas.Archive], summary="Fetch latest URL archives for the authenticated user.")
def latest(skip: int = 0, limit: int = 25, db: Session = Depends(get_db_dependency), email=Depends(get_user_auth)):
@url_router.get("/latest", summary="Fetch latest URL archives for the authenticated user.")
def latest(skip: int = 0, limit: int = 25, db: Session = Depends(get_db_dependency), email=Depends(get_user_auth)) -> list[schemas.ArchiveResult]:
return crud.search_archives_by_email(db, email, skip=skip, limit=limit)
@url_router.get("/{id}", response_model=schemas.Archive, summary="Fetch a single URL archive by the associated id.")
def lookup(id, db: Session = Depends(get_db_dependency), email=Depends(get_token_or_user_auth)):
@url_router.get("/{id}", summary="Fetch a single URL archive by the associated id.")
def lookup(id, db: Session = Depends(get_db_dependency), email=Depends(get_token_or_user_auth)) -> schemas.ArchiveResult:
archive = crud.get_archive(db, id, email)
if archive is None:
raise HTTPException(status_code=404, detail="Archive not found")
return archive
@url_router.delete("/{id}", response_model=schemas.TaskDelete, summary="Delete a single URL archive by id.")
def delete_task(id, db: Session = Depends(get_db_dependency), email=Depends(get_user_auth)):
@url_router.delete("/{id}", summary="Delete a single URL archive by id.")
def delete_task(id, db: Session = Depends(get_db_dependency), email=Depends(get_user_auth)) -> schemas.TaskDelete:
logger.info(f"deleting url archive task {id} request by {email}")
return JSONResponse({
"id": id,