refactor: task builder has become redundant

task creation has been simplified to the point where a factory was pointless

Author: toresbe

app/api/hooks/metadata.py

@@ -7,7 +7,7 @@
 
 from app.api.hooks.schema.request import HookRequest
 from app.media.ffprobe_schema import FfprobeOutput
-from app.task_builder import build_task
+from app.runner import Task
 
 logger = logging.getLogger(__name__)
 
@@ -33,8 +33,8 @@ class MetadataExtractor:
     """Class to handle metadata extraction and compliance checking."""
 
     async def _run_ffprobe(self, filepath: Path) -> str:
-        stdout, _ = await build_task(
-            f"ffprobe -v quiet -show_format -show_streams -of json {shlex.quote(str(filepath))}"
+        stdout, _ = await Task(
+            f"ffprobe -v quiet -show_format -show_streams -of json {shlex.quote(str(filepath))}",
         ).execute()
         return stdout
 

app/ingest.py

@@ -8,7 +8,7 @@
 from .archive_store import Archive
 from .media.comand_template import ProfileTemplateArguments, TemplatedCommandGenerator
 from .media.ffprobe_schema import FfprobeOutput
-from .task_builder import build_task
+from .runner import Task
 
 DESIRED_FORMATS = (
     FormatEnum.LARGE_THUMB,
@@ -90,7 +90,7 @@ async def _process_format(
         command = template.render(template_args)
         self.logger.debug("Generated command: %s", command)
 
-        await build_task(command).execute()
+        await Task(command).execute()
 
         self.logger.info("Creating video file entry for %s", output_file)
         await self.django_api.create_video_file(filename=str(output_file), file_format=file_format, video_id=video_id)

app/runner.py

@@ -1,7 +1,5 @@
 import logging
-from asyncio.subprocess import Process
-from collections.abc import Coroutine
-from typing import Any
+from asyncio import subprocess
 
 logger = logging.getLogger(__name__)
 
@@ -11,10 +9,12 @@ class Task:
     Raises ChildProcessError if the command fails.
     """
 
-    proc: Coroutine[Any, Any, Process]
+    command_line: str
 
-    def __init__(self, proc: Coroutine[Any, Any, Process]):
-        self.proc = proc
+    def __init__(self, command_line: str):
+        self.command_line = command_line
+        self.proc = subprocess.create_subprocess_shell(command_line, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        logger.debug("Created task for command: %s", command_line)
 
     async def execute(self) -> tuple[str, str]:
         proc = await self.proc

app/task_builder.py

@@ -1,6 +1,6 @@
 import pytest
 
-from app.task_builder import build_task
+from app.runner import Task
 
 
 @pytest.mark.asyncio
@@ -9,6 +9,6 @@ async def test_basic_run(tmp_path):
     test_path.touch()
     assert test_path.exists()
 
-    await build_task(f"rm {test_path}").execute()
+    await Task(f"rm {test_path}").execute()
 
     assert not test_path.exists()