secure media endpoints

kontor-api/src/apis/version1/mediafile.py

@@ -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])
+@router.get(
-# def get_all_files(db: SessionDep, review: bool = False, download: bool = False, current_user: Profile = Depends(get_current_user_from_token)) -> List[MediaFileResponse]:
+    "/files",
-def get_all_files(db: SessionDep, review: bool = False, download: bool = False) -> list[MediaFileResponse]: # type: ignore
+    response_model=list[MediaFileResponse],
-    results: list[MediaFileResponse] = []
+    dependencies=[Depends(get_current_user_from_token)],
-    files: Sequence[MediaFile]
+)
+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
+        files = db.query(MediaFile).all()
-    for mediafile in files: # type: ignore
+    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)

kontor-api/src/core/security.py

@@ -13,13 +13,23 @@ from pydantic import ValidationError
 from src.core.config import settings
 from src.core.log_conf import logger
 from src.db.models.admin import Profile
-from src.db.repository.admin import get_profile_by_username, get_profile_by_email, is_database_empty
+from src.db.repository.admin import (
+    get_profile_by_username,
+    get_profile_by_email,
+    is_database_empty,
+)
 from src.db.session import SessionLocal
 from src.schema.admin import ProfileModel, TokenData
 
 oauth2_scheme = OAuth2PasswordBearer(
     tokenUrl="/token",
-    scopes={"me": "read", "admin": "read", "ROLE_ADMIN": "admin", "ROLE_MEDIA": "media", "ROLE_USER": "user"},
+    scopes={
+        "me": "read",
+        "admin": "read",
+        "ROLE_ADMIN": "admin",
+        "ROLE_MEDIA": "media",
+        "ROLE_USER": "user",
+    },
 )
 
 
@@ -51,6 +61,7 @@ oauth2_scheme = OAuth2PasswordBearer(
 #                 return None
 #         return param
 
+
 def authenticate_user_by_email(email: str, password: str) -> Optional[Profile]:
     with SessionLocal() as db:
         user = get_profile_by_email(email=email, db=db)
@@ -161,6 +172,7 @@ async def get_current_active_user(
 
 
 def get_current_user_from_token(token: str = Depends(oauth2_scheme)):
+    """ """
     credentials_exception = HTTPException(
         status_code=status.HTTP_401_UNAUTHORIZED,
         detail="Could not validate credentials",
@@ -173,10 +185,13 @@ def get_current_user_from_token(token: str = Depends(oauth2_scheme)):
         logger.info("username/email extracted is %s", username)
         if username is None:
             raise credentials_exception
-    except JWTError:
+    except JWTError as exception:
-        raise credentials_exception
+        raise credentials_exception from exception
     with SessionLocal() as db:
         user = get_profile_by_email(email=username, db=db)
     if user is None:
         raise credentials_exception
     return user
+
+
+UserDep = Annotated[Profile, Depends(get_current_user_from_token)]

kontor-api/src/db/models/media.py

@@ -14,51 +14,71 @@ 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'
+    __tablename__ = "media_actor"
     name: Mapped[str]
     url: Mapped[Optional[str]] = mapped_column(unique=True)
     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'
+    __tablename__ = "media_actor_file"
-    media_actor_id: Mapped[str] = mapped_column(ForeignKey("media_actor.id"), nullable=False)
+    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]
@@ -68,10 +88,10 @@ 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})'
+            return f"{self.title}({self.id})"

kontor-api/src/db/session.py

@@ -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)]