secure media endpoints

2026-05-17 21:48:40 +02:00
parent f1abf03b35
commit b2cf1c4698
3 changed files with 116 additions and 44 deletions
@@ -1,7 +1,14 @@
+from typing import List
+
 from fastapi import APIRouter, status, HTTPException, Depends
 from sqlalchemy import select, Sequence
 from src.core.log_conf import logger
-from src.db.repository.media import create_new_mediaactorfile, create_new_mediafile, delete_mediafile
+from src.core.security import UserDep, get_current_user_from_token
+from src.db.repository.media import (
+    create_new_mediaactorfile,
+    create_new_mediafile,
+    delete_mediafile,
+)
 from src.db.session import SessionDep
 from src.schema.media.actor import MediaActorResponse
 from src.schema.media.actorfile import MediaActorFileResponse
@@ -10,10 +17,14 @@ from src.db.models.media import MediaFile

 router = APIRouter()

+
@router.get("/update-titles")
-def update_titles(db: SessionDep) -> list[MediaFileResponse]: # type: ignore
+def update_titles(db: SessionDep) -> list[MediaFileResponse]:
+    """
+    Update title for given MediaFile.
+    """
    results: list[MediaFileResponse] = []
-    files = db.query(MediaFile).filter(MediaFile.review == True).all()
+    files = db.query(MediaFile).filter(MediaFile.review.is_(True)).all()
    for mediafile in files:
        mediafile.update_title()
        db.add(mediafile)
@@ -23,61 +34,92 @@ def update_titles(db: SessionDep) -> list[MediaFileResponse]: # type: ignore
    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]: # type: ignore
-    results: list[MediaFileResponse] = []
-    files: Sequence[MediaFile]
+@router.get(
+    "/files",
+    response_model=list[MediaFileResponse],
+    dependencies=[Depends(get_current_user_from_token)],
+)
+def get_all_files(
+    db: SessionDep, review: bool = False, download: bool = False
+) -> List[MediaFileResponse]:
+    """
+    Get all MediaFiles.
+    """
+    results: List[MediaFileResponse] = []
+    files: List[MediaFile]
    if review:
-        files = db.query(MediaFile).filter(MediaFile.review == True).all() # type: ignore
+        files = db.query(MediaFile).filter(MediaFile.review.is_(True)).all()
    elif download:
-        files = db.query(MediaFile).filter(MediaFile.should_download == True).all() # type: ignore
+        files = db.query(MediaFile).filter(MediaFile.should_download.is_(True)).all()
    else:
-        files = db.scalars(select(MediaFile)).all() # type: ignore
-    for mediafile in files: # type: ignore
+        files = db.query(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: str, db: SessionDep) -> MediaFileResponse: # type: ignore
+
+@router.get(
+    "/files/{file_id}",
+    response_model=MediaFileResponse,
+    dependencies=[Depends(get_current_user_from_token)],
+)
+def get_file(file_id: str, db: SessionDep) -> MediaFileResponse:
+    """
+    Get MediaFile with given id or return HTTPException.
+    """
    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.delete("/files/{file_id}", status_code=status.HTTP_204_NO_CONTENT)
-def delete_file(file_id: str, db: SessionDep): # type: ignore
+def delete_file(file_id: str, db: SessionDep):
+    """
+    Delete MediaFile by given id.
+    """
    mediafile = db.get(MediaFile, file_id)
    if not mediafile:
        raise HTTPException(status_code=404, detail="MediaFile could not be found")
-    logger.info(f"delete MediaFile: {file_id}")
+    logger.info("delete MediaFile: %s", file_id)
    actor_files = mediafile.media_actor_files
-    logger.info(f"MediaActorFiles links {len(actor_files)}")
+    logger.info("MediaActorFiles links %s", len(actor_files))
    if len(actor_files) == 0:
        delete_mediafile(db, mediafile.id)

+
@router.get("/files/{file_id}/actors", response_model=list[MediaActorResponse])
-def get_file_actors(file_id: str, db: SessionDep) -> list[MediaActorResponse]: # type: ignore
+def get_file_actors(file_id: str, db: SessionDep) -> list[MediaActorResponse]:
+    """
+    Get list of ACtors for given MediaFile.
+    """
    mediafile = db.get(MediaFile, file_id)
    if not mediafile:
        raise HTTPException(status_code=404, detail="MediaFile could not be found")
    actor_files = mediafile.media_actor_files
-    logger.info(f"already known actors: {actor_files}")
+    logger.info("already known actors: %s", actor_files)
    results: list[MediaActorResponse] = []
    for actor_file in actor_files:
-        response = MediaActorResponse(id=actor_file.media_actor.id, name=actor_file.media_actor.name, url=actor_file.media_actor.url)
+        response = MediaActorResponse(
+            id=actor_file.media_actor.id,
+            name=actor_file.media_actor.name,
+            url=str(actor_file.media_actor.url),
+        )
        results.append(response)
    return results

+
@router.put("/files/{file_id}/actors", response_model=list[MediaActorFileResponse])
-def update_file_actors(file_id: str, db: SessionDep, actors: list[MediaActorResponse]) -> list[MediaActorFileResponse]: # type: ignore
+def update_file_actors(
+    file_id: str, db: SessionDep, actors: list[MediaActorResponse]
+) -> list[MediaActorFileResponse]:  # type: ignore
    mediafile = db.get(MediaFile, file_id)
    if not mediafile:
        raise HTTPException(status_code=404, detail="MediaFile could not be found")
    actor_files = mediafile.media_actor_files
-    logger.info(f"already known actors: {actor_files}")
+    logger.info("already known actors: %s", actor_files)
    for actor in actors:
        already_associated = False
        for actor_file in actor_files:
@@ -91,12 +133,19 @@ def update_file_actors(file_id: str, db: SessionDep, actors: list[MediaActorResp
    actor_files = mediafile.media_actor_files
    results: list[MediaActorFileResponse] = []
    for actor_file in actor_files:
-        response = MediaActorFileResponse(id=actor_file.id, actor_id=actor_file.media_actor_id, file_id=actor_file.media_file_id)
+        response = MediaActorFileResponse(
+            id=actor_file.id,
+            actor_id=actor_file.media_actor_id,
+            file_id=actor_file.media_file_id,
+        )
        results.append(response)
    return results

+
@router.put("/files/{file_id}", response_model=MediaFileResponse)
-def update_file(file_id: str, db: SessionDep, info: MediaFileResponse) -> MediaFileResponse: # type: ignore
+def update_file(
+    file_id: str, db: SessionDep, info: MediaFileResponse
+) -> MediaFileResponse:  # type: ignore
    mediaFile = db.get(MediaFile, file_id)
    if not mediaFile:
        raise HTTPException(status_code=404, detail="MediaFile could not be found")
@@ -109,8 +158,9 @@ def update_file(file_id: str, db: SessionDep, info: MediaFileResponse) -> MediaF
    response = get_file_details(mediafile)
    return response

+
@router.post("/files", status_code=status.HTTP_201_CREATED)
-def add_file(new_link: Link, db: SessionDep) -> MediaFileResponse: # type: ignore
+def add_file(new_link: Link, db: SessionDep) -> MediaFileResponse:  # type: ignore
    logger.info(f"add url {new_link.url}")
    try:
        mediaFile: MediaFile = create_new_mediafile(new_link.url, db)
@@ -14,50 +14,70 @@ from src.db.models.base import Base, BaseMixin, BaseVideoMixin


 class MediaFile(Base, BaseMixin, BaseVideoMixin):
-    __tablename__ = 'media_file'
-    media_actor_files: Mapped[List["MediaActorFile"]] = relationship(back_populates="media_file")
+    """
+    MediaFile represents video link.
+    """
+
+    __tablename__ = "media_file"
+    media_actor_files: Mapped[List["MediaActorFile"]] = relationship(
+        back_populates="media_file"
+    )

    def __repr__(self):
-        return f'MediaFile({self.id} {self.title} {self.title})'
+        return f"MediaFile({self.id} {self.title} {self.title})"

    def __str__(self):
-        return f'{self.title}({self.id})'
+        return f"{self.title}({self.id})"
+
+    def update_title(self):
+        """
+        Update title from url.
+        """


 class MediaActor(Base, BaseMixin):
    __tablename__ = 'media_actor'
    name = Column(String)
    media_actor_files = relationship("MediaActorFile")
-    
+
    def __repr__(self) -> str:
-        return f'MediaActor({self.id} {self.name} {self.url})'
-    
+        return f"MediaActor({self.id} {self.name} {self.url})"
+
    def __str__(self) -> str:
-        return f'{self.url}({self.id})'
+        return f"{self.url}({self.id})"


 class MediaActorFile(Base, BaseMixin):
-    __tablename__ = 'media_actor_file'
-    media_actor_id: Mapped[str] = mapped_column(ForeignKey("media_actor.id"), nullable=False)
+    __tablename__ = "media_actor_file"
+    media_actor_id: Mapped[str] = mapped_column(
+        ForeignKey("media_actor.id"), nullable=False
+    )
    media_actor: Mapped[MediaActor] = relationship(back_populates="media_actor_files")
-    media_file_id: Mapped[str] = mapped_column(ForeignKey("media_file.id"), nullable=True)
+    media_file_id: Mapped[str] = mapped_column(
+        ForeignKey("media_file.id"), nullable=True
+    )
    media_file: Mapped[MediaFile] = relationship(back_populates="media_actor_files")

    def __repr__(self):
-        return f'MediaActorFile({self.id} {self.media_actor_id} {self.media_file_id})'
+        return f"MediaActorFile({self.id} {self.media_actor_id} {self.media_file_id})"

    def __str__(self) -> str:
-        return f'{self.id} {self.media_actor_id} {self.media_file_id}'
+        return f"{self.id} {self.media_actor_id} {self.media_file_id}"
+

 class MediaArticle(Base, BaseMixin):
-    __tablename__ = 'media_article'
+    __tablename__ = "media_article"
    review: Mapped[bool]
    title: Mapped[str]
    url: Mapped[str] = mapped_column(unique=True)


 class MediaVideo(Base, BaseMixin):
-    __tablename__ = 'media_video'
+    """
+    MediaFile represents video link.
+    """
+
+    __tablename__ = "media_video"
    cloud_link: Mapped[str]
    file_name: Mapped[str]
    path: Mapped[str]
@@ -67,11 +87,11 @@ class MediaVideo(Base, BaseMixin):
    should_download: Mapped[bool]

    def __repr__(self):
-        return f'MediaFile({self.id} {self.title} {self.url})'
+        return f"MediaFile({self.id} {self.title} {self.url})"

    def __str__(self):
        if self.title is None:
-            return f'{self.url}({self.id})'
+            return f"{self.url}({self.id})"
        else:
            return f'{self.title}({self.id})'

@@ -12,8 +12,10 @@ engine = create_engine(SQLALCHEMY_DATABASE_URL)
 SessionLocal = sessionmaker(bind=engine)


-def get_db() -> Generator:
+def get_db() -> Generator[Session, None, None]:
+    """ """
    with SessionLocal() as db:
        yield db

-SessionDep: type[Session] = Annotated[Session, Depends(get_db)]
+
+SessionDep = Annotated[Session, Depends(get_db)]