chore(api): seperate out core files

Signed-off-by: Jordan Shatford <[email protected]>

apps/api/app/core/managers.py

@@ -0,0 +1,44 @@
+from typing import Callable
+
+from ..models import DownloadStatusUpdate
+from ..models import VideoWithOptions
+from ..models import VideoWithOptionsAndStatus
+from .threads import YoutubeDownloadThread
+
+
+class AudioDownloadManager:
+    def __init__(
+        self, output_dir: str,
+        status_hook: Callable[[DownloadStatusUpdate], None],
+    ):
+        self._status_hook = status_hook
+        self._downloads: dict[str, YoutubeDownloadThread] = {}
+        self._output_dir = output_dir
+
+    def __contains__(self, video_id: str) -> bool:
+        return video_id in self._downloads
+
+    def add(self, video: VideoWithOptions) -> None:
+        if video.id not in self._downloads:
+            download = YoutubeDownloadThread(
+                video, self._output_dir, self.send_status_update,
+            )
+            self._downloads[video.id] = download
+            download.start()
+
+    def remove(self, video_id: str) -> None:
+        if video_id in self._downloads:
+            self._downloads[video_id].remove()
+            self._downloads.pop(video_id, None)
+
+    def get(self, video_id: str) -> YoutubeDownloadThread | None:
+        return self._downloads.get(video_id, None)
+
+    def get_all_videos(self) -> list[VideoWithOptionsAndStatus]:
+        return [
+            VideoWithOptionsAndStatus(**d.video.dict(), status=d.status)
+            for d in self._downloads.values()
+        ]
+
+    def send_status_update(self, update: DownloadStatusUpdate) -> None:
+        self._status_hook(update)

apps/api/app/dependencies.py

@@ -8,9 +8,9 @@
 from fastapi.security import HTTPAuthorizationCredentials
 from fastapi.security import HTTPBearer
 
-from .utils.managers import Session
-from .utils.managers import session_manager
-from .utils.threads import YoutubeDownloadThread
+from .core.threads import YoutubeDownloadThread
+from .session import Session
+from .session import session_manager
 
 AdditionResponses: TypeAlias = dict[int | str, dict[str, Any]]
 

apps/api/app/main.py

@@ -9,7 +9,7 @@
 from .routers import downloads
 from .routers import search
 from .routers import session
-from .utils.managers import session_manager
+from .session import session_manager
 
 # Read data from pyproject.toml file
 pyproject_data: dict[str, Any] = {}

apps/api/app/routers/downloads.py

@@ -16,8 +16,8 @@
 from ..models import DownloadStatusUpdate
 from ..models import VideoWithOptions
 from ..models import VideoWithOptionsAndStatus
-from ..utils.managers import Session
-from ..utils.managers import session_manager
+from ..session import Session
+from ..session import session_manager
 
 router = APIRouter(
     prefix='/downloads',

apps/api/app/routers/search.py

@@ -2,10 +2,10 @@
 from fastapi import HTTPException
 from fastapi import status
 
+from ..core.youtube import search_youtube
 from ..dependencies import depends_session_responses
 from ..dependencies import DependsSession
 from ..models import Video
-from ..utils.youtube import search_youtube
 
 router = APIRouter(
     prefix='/search',

apps/api/app/routers/session.py

@@ -4,7 +4,7 @@
 from ..dependencies import depends_session_responses
 from ..dependencies import DependsSession
 from ..models import Session
-from ..utils.managers import session_manager
+from ..session import session_manager
 
 router = APIRouter(
     prefix='/session',

apps/api/app/utils/managers.py → apps/api/app/session.py

@@ -4,51 +4,10 @@
 from datetime import datetime
 from datetime import timedelta
 from queue import Queue
-from typing import Callable
 
-from ..models import DownloadStatusUpdate
-from ..models import VideoWithOptions
-from ..models import VideoWithOptionsAndStatus
-from .threads import RepeatedTimer
-from .threads import YoutubeDownloadThread
-
-
-class AudioDownloadManager:
-    def __init__(
-        self, output_dir: str,
-        status_hook: Callable[[DownloadStatusUpdate], None],
-    ):
-        self._status_hook = status_hook
-        self._downloads: dict[str, YoutubeDownloadThread] = {}
-        self._output_dir = output_dir
-
-    def __contains__(self, video_id: str) -> bool:
-        return video_id in self._downloads
-
-    def add(self, video: VideoWithOptions) -> None:
-        if video.id not in self._downloads:
-            download = YoutubeDownloadThread(
-                video, self._output_dir, self.send_status_update,
-            )
-            self._downloads[video.id] = download
-            download.start()
-
-    def remove(self, video_id: str) -> None:
-        if video_id in self._downloads:
-            self._downloads[video_id].remove()
-            self._downloads.pop(video_id, None)
-
-    def get(self, video_id: str) -> YoutubeDownloadThread | None:
-        return self._downloads.get(video_id, None)
-
-    def get_all_videos(self) -> list[VideoWithOptionsAndStatus]:
-        return [
-            VideoWithOptionsAndStatus(**d.video.dict(), status=d.status)
-            for d in self._downloads.values()
-        ]
-
-    def send_status_update(self, update: DownloadStatusUpdate) -> None:
-        self._status_hook(update)
+from .core.managers import AudioDownloadManager
+from .core.threads import RepeatedTimer
+from .models import DownloadStatusUpdate
 
 
 class Session: