Thomas Peetz
70 lines
2.7 KiB
Python
70 lines
2.7 KiB
Python
from typing import List, AnyStr
|
|
|
|
from fastapi import APIRouter, status, HTTPException, Depends
|
|
from sqlalchemy import select, Sequence
|
|
from src.core.log_conf import logger
|
|
from src.apis.utils import SessionDep
|
|
from src.db.repository.media import create_new_mediafile
|
|
from src.schema.media.file import MediaFileResponse, Link, get_file_details, set_file
|
|
from src.db.models.media import MediaFile
|
|
|
|
router = APIRouter()
|
|
|
|
@router.get("/update-titles")
|
|
def update_titles(db: SessionDep) -> list[MediaFileResponse]:
|
|
results: list[MediaFileResponse] = []
|
|
files = db.query(MediaFile).filter(MediaFile.review == True).all()
|
|
for mediafile in files:
|
|
mediafile.update_title()
|
|
db.add(mediafile)
|
|
response = get_file_details(mediafile)
|
|
results.append(response)
|
|
db.commit()
|
|
return results
|
|
|
|
|
|
@router.get("/files", response_model=List[MediaFileResponse])
|
|
#def get_all_files(db: SessionDep, review: bool = False, download: bool = False, current_user: Profile = Depends(get_current_user_from_token)) -> List[MediaFileResponse]:
|
|
def get_all_files(db: SessionDep, review: bool = False, download: bool = False) -> List[MediaFileResponse]:
|
|
results: list[MediaFileResponse] = []
|
|
files: Sequence[MediaFile]
|
|
if review:
|
|
files = db.query(MediaFile).filter(MediaFile.review == True).all()
|
|
elif download:
|
|
files = db.query(MediaFile).filter(MediaFile.should_download == True).all()
|
|
else:
|
|
files = db.scalars(select(MediaFile)).all()
|
|
for mediafile in files:
|
|
response = get_file_details(mediafile)
|
|
results.append(response)
|
|
return results
|
|
|
|
@router.get("/files/{file_id}", response_model=MediaFileResponse)
|
|
def get_file(file_id: AnyStr, db: SessionDep) -> MediaFileResponse:
|
|
mediafile = db.get(MediaFile, file_id)
|
|
if not mediafile:
|
|
raise HTTPException(status_code=404, detail="MediaFile could not be found")
|
|
response = get_file_details(mediafile)
|
|
return response
|
|
|
|
@router.put("/files/{file_id}", response_model=MediaFileResponse)
|
|
def update_file(file_id: AnyStr, db: SessionDep, info: MediaFileResponse) -> MediaFileResponse:
|
|
mediaFile = db.get(MediaFile, file_id)
|
|
if not mediaFile:
|
|
raise HTTPException(status_code=404, detail="MediaFile could not be found")
|
|
set_file(info, mediaFile)
|
|
db.add(mediaFile)
|
|
db.commit()
|
|
return info
|
|
|
|
|
|
@router.post("/files", status_code=status.HTTP_201_CREATED)
|
|
def add_file(new_link: Link, db: SessionDep) -> MediaFileResponse:
|
|
logger.info(f"add url {new_link.url}")
|
|
try:
|
|
mediaFile: MediaFile = create_new_mediafile(new_link.url, db)
|
|
except:
|
|
raise HTTPException(status_code=409, detail="Link duplicate")
|
|
response = get_file_details(mediaFile)
|
|
return response
|