create new file specific to status

sidecar/app/query/base.py

@@ -1,18 +1,17 @@
-# pylint: disable=R0801
 from __future__ import annotations
 
-import time
 from collections.abc import Iterable
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import ClassVar, NamedTuple, Optional, TypeVar
+from typing import ClassVar, Optional, TypeVar
 
 import loguru
 
 from ..helpers import Role
 from ..logger import logger
 from ..settings import settings
-from .step import Status, Step
+from .status import Status, StatusHistory
+from .step import Step
 
 # Dictionary to store queries
 queries: dict[str, "Query"] = {}
@@ -22,76 +21,18 @@ class QueryExistsError(Exception):
     pass
 
 
-StatusChangeEvent = NamedTuple(
-    "StatusChangeEvent", [("status", Status), ("timestamp", float)]
-)
-
-
-@dataclass
-class StatusHistory:
-    file_path: Path = field(init=True, repr=False)
-    logger: loguru.Logger = field(init=True, repr=False)
-    _status_history: list[StatusChangeEvent] = field(
-        init=False, default_factory=list, repr=True
-    )
-
-    def __post_init__(self):
-        if self.file_path.exists():
-            self.logger.debug(f"Loading status history from file {self.file_path}")
-            with self.file_path.open("r", encoding="utf8") as f:
-                for line in f:
-                    status_str, timestamp = line.split(",")
-                    self._status_history.append(
-                        StatusChangeEvent(
-                            status=Status[status_str], timestamp=float(timestamp)
-                        )
-                    )
-
-    def add(self, status: Status, timestamp: float = time.time()):
-        self._status_history.append(
-            StatusChangeEvent(status=status, timestamp=timestamp)
-        )
-        with self.file_path.open("a", encoding="utf8") as f:
-            self.logger.debug(f"updating status: {status=}")
-            f.write(f"{status.name},{timestamp}\n")
-
-    @property
-    def last_status_event(self):
-        if not self._status_history:
-            return StatusChangeEvent(status=Status.UNKNOWN, timestamp=time.time())
-        return self._status_history[-1]
-
-    @property
-    def current_status(self):
-        return self.last_status_event.status
-
-    @property
-    def status_event_json(self):
-        status_event = {
-            "status": self.last_status_event.status.name,
-            "start_time": self.last_status_event.timestamp,
-        }
-        if self.current_status >= Status.COMPLETE and len(self._status_history) >= 2:
-            status_event["start_time"] = self._status_history[-2].timestamp
-            status_event["end_time"] = self.last_status_event.timestamp
-        return status_event
-
-
 @dataclass
 class Query:
-    # pylint: disable=too-many-instance-attributes
     query_id: str
     current_step: Optional[Step] = field(init=False, default=None, repr=True)
     logger: loguru.Logger = field(init=False, repr=False)
     _logger_id: int = field(init=False, repr=False)
-    _log_dir: Path = settings.root_path / Path("logs")
     _status_history: StatusHistory = field(init=False, repr=True)
-    _status_dir: Path = settings.root_path / Path("status_semaphore")
     step_classes: ClassVar[list[type[Step]]] = []
 
     def __post_init__(self):
         self.logger = logger.bind(task=self.query_id)
-        status_dir = settings.root_path / Path("status_semaphore")
+        status_dir = settings.root_path / Path("status")
         status_dir.mkdir(exist_ok=True)
         status_file_path = status_dir / Path(f"{self.query_id}")
         self._status_history = StatusHistory(file_path=status_file_path, logger=logger)
@@ -108,6 +49,10 @@ def __post_init__(self):
             raise QueryExistsError(f"{self.query_id} already exists")
         queries[self.query_id] = self
 
+    @property
+    def _log_dir(self) -> Path:
+        return settings.root_path / Path("logs")
+
     @property
     def role(self) -> Role:
         return settings.role

sidecar/app/query/status.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+from enum import IntEnum, auto
+from pathlib import Path
+from typing import NamedTuple
+
+import loguru
+
+
+class Status(IntEnum):
+    UNKNOWN = auto()
+    STARTING = auto()
+    COMPILING = auto()
+    WAITING_TO_START = auto()
+    IN_PROGRESS = auto()
+    COMPLETE = auto()
+    KILLED = auto()
+    NOT_FOUND = auto()
+    CRASHED = auto()
+
+
+StatusChangeEvent = NamedTuple(
+    "StatusChangeEvent", [("status", Status), ("timestamp", float)]
+)
+
+
+@dataclass
+class StatusHistory:
+    file_path: Path = field(init=True, repr=False)
+    logger: loguru.Logger = field(init=True, repr=False)
+    _status_history: list[StatusChangeEvent] = field(
+        init=False, default_factory=list, repr=True
+    )
+
+    def __post_init__(self):
+        if self.file_path.exists():
+            self.logger.debug(f"Loading status history from file {self.file_path}")
+            with self.file_path.open("r", encoding="utf8") as f:
+                for line in f:
+                    status_str, timestamp = line.split(",")
+                    self._status_history.append(
+                        StatusChangeEvent(
+                            status=Status[status_str], timestamp=float(timestamp)
+                        )
+                    )
+
+    def add(self, status: Status, timestamp: float = time.time()):
+        self._status_history.append(
+            StatusChangeEvent(status=status, timestamp=timestamp)
+        )
+        with self.file_path.open("a", encoding="utf8") as f:
+            self.logger.debug(f"updating status: {status=}")
+            f.write(f"{status.name},{timestamp}\n")
+
+    @property
+    def last_status_event(self):
+        if not self._status_history:
+            return StatusChangeEvent(status=Status.UNKNOWN, timestamp=time.time())
+        return self._status_history[-1]
+
+    @property
+    def current_status(self):
+        return self.last_status_event.status
+
+    @property
+    def status_event_json(self):
+        status_event = {
+            "status": self.last_status_event.status.name,
+            "start_time": self.last_status_event.timestamp,
+        }
+        if self.current_status >= Status.COMPLETE and len(self._status_history) >= 2:
+            status_event["start_time"] = self._status_history[-2].timestamp
+            status_event["end_time"] = self.last_status_event.timestamp
+        return status_event

sidecar/app/query/step.py

@@ -3,29 +3,17 @@
 import os
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
-from enum import IntEnum, auto
 from typing import TYPE_CHECKING, ClassVar, Optional
 
 import loguru
 
 from .command import Command
+from .status import Status
 
 if TYPE_CHECKING:
     from .base import QueryTypeT
 
 
-class Status(IntEnum):
-    UNKNOWN = auto()
-    STARTING = auto()
-    COMPILING = auto()
-    WAITING_TO_START = auto()
-    IN_PROGRESS = auto()
-    COMPLETE = auto()
-    KILLED = auto()
-    NOT_FOUND = auto()
-    CRASHED = auto()
-
-
 @dataclass(kw_only=True)
 class Step(ABC):
     skip: bool = field(init=False, default=False)

sidecar/app/routes/start.py

@@ -10,7 +10,7 @@
 from ..query.base import Query
 from ..query.demo_logger import DemoLoggerQuery
 from ..query.ipa import GateType, IPACoordinatorQuery, IPAHelperQuery
-from ..query.step import Status
+from ..query.status import Status
 from ..settings import settings
 
 router = APIRouter(

sidecar/app/routes/stop.py

@@ -2,7 +2,7 @@
 
 from ..logger import logger
 from ..query.base import Query
-from ..query.step import Status
+from ..query.status import Status
 
 router = APIRouter(
     prefix="/stop",

sidecar/app/routes/websockets.py

@@ -7,7 +7,7 @@
 
 from ..logger import logger
 from ..query.base import Query
-from ..query.step import Status
+from ..query.status import Status
 
 router = APIRouter(
     prefix="/ws",