Move actions and handlers into preset fine_logs

Author: Aksem

presets/fine_logs/fine_logs/__init__.py

@@ -0,0 +1,21 @@
+from fine_logs.list_observability_services_action import ListObservabilityServicesAction
+from fine_logs.get_service_logs_action import GetServiceLogsAction
+from fine_logs.clean_service_logs_action import CleanServiceLogsAction
+from fine_logs.clean_services_logs_action import CleanServicesLogsAction
+from fine_logs.list_observability_services_handler import ListObservabilityServicesHandler
+from fine_logs.get_service_logs_handler import GetServiceLogsHandler
+from fine_logs.clean_service_logs_handler import CleanServiceLogsHandler
+from fine_logs.clean_services_logs_discovery_handler import CleanServicesLogsDiscoveryHandler
+from fine_logs.clean_services_logs_iterate_handler import CleanServicesLogsIterateHandler
+
+__all__ = [
+    "ListObservabilityServicesAction",
+    "GetServiceLogsAction",
+    "CleanServiceLogsAction",
+    "CleanServicesLogsAction",
+    "ListObservabilityServicesHandler",
+    "GetServiceLogsHandler",
+    "CleanServiceLogsHandler",
+    "CleanServicesLogsDiscoveryHandler",
+    "CleanServicesLogsIterateHandler",
+]

presets/fine_logs/fine_logs/clean_service_logs_action.py

@@ -0,0 +1,53 @@
+# docs: docs/reference/actions.md
+import dataclasses
+
+from finecode_extension_api import code_action, textstyler
+
+
+@dataclasses.dataclass
+class CleanServiceLogsRunPayload(code_action.RunActionPayload):
+    service_id: str
+    """Service whose logs should be deleted, as returned by list_observability_services."""
+
+
+class CleanServiceLogsRunContext(
+    code_action.RunActionContext[CleanServiceLogsRunPayload]
+): ...
+
+
+@dataclasses.dataclass
+class CleanServiceLogsRunResult(code_action.RunActionResult):
+    errors: list[str] = dataclasses.field(default_factory=list)
+
+    def update(self, other: code_action.RunActionResult) -> None:
+        if not isinstance(other, CleanServiceLogsRunResult):
+            return
+        self.errors += other.errors
+
+    def to_text(self) -> str | textstyler.StyledText:
+        if self.errors:
+            return "\n".join(self.errors)
+        return "Logs cleaned."
+
+    @property
+    def return_code(self) -> code_action.RunReturnCode:
+        return (
+            code_action.RunReturnCode.ERROR
+            if self.errors
+            else code_action.RunReturnCode.SUCCESS
+        )
+
+
+class CleanServiceLogsAction(
+    code_action.Action[
+        CleanServiceLogsRunPayload,
+        CleanServiceLogsRunContext,
+        CleanServiceLogsRunResult,
+    ]
+):
+    """Delete all logs for a specific observability service."""
+
+    DESCRIPTION = "Delete all logs for a specific observability service."
+    PAYLOAD_TYPE = CleanServiceLogsRunPayload
+    RUN_CONTEXT_TYPE = CleanServiceLogsRunContext
+    RESULT_TYPE = CleanServiceLogsRunResult

presets/fine_logs/fine_logs/clean_service_logs_handler.py

@@ -0,0 +1,53 @@
+import dataclasses
+
+from finecode_extension_api import code_action
+from fine_logs.clean_service_logs_action import (
+    CleanServiceLogsAction,
+    CleanServiceLogsRunContext,
+    CleanServiceLogsRunPayload,
+    CleanServiceLogsRunResult,
+)
+from finecode_extension_api.interfaces import (
+    iextensionrunnerinfoprovider,
+    ilogger,
+)
+
+from fine_logs.observability_log_utils import resolve_log_dir
+
+
+@dataclasses.dataclass
+class CleanServiceLogsHandlerConfig(code_action.ActionHandlerConfig): ...
+
+
+class CleanServiceLogsHandler(
+    code_action.ActionHandler[
+        CleanServiceLogsAction,
+        CleanServiceLogsHandlerConfig,
+    ]
+):
+    def __init__(
+        self,
+        runner_info_provider: iextensionrunnerinfoprovider.IExtensionRunnerInfoProvider,
+        logger: ilogger.ILogger,
+    ) -> None:
+        self.runner_info_provider = runner_info_provider
+        self.logger = logger
+
+    async def run(
+        self,
+        payload: CleanServiceLogsRunPayload,
+        run_context: CleanServiceLogsRunContext,
+    ) -> CleanServiceLogsRunResult:
+        log_dir = resolve_log_dir(payload.service_id, self.runner_info_provider)
+        if not log_dir.is_dir():
+            return CleanServiceLogsRunResult()
+
+        errors: list[str] = []
+        for log_file in log_dir.glob("*.log"):
+            try:
+                log_file.unlink()
+                self.logger.info(f"Deleted {log_file}")
+            except Exception as e:
+                errors.append(str(e))
+
+        return CleanServiceLogsRunResult(errors=errors)

presets/fine_logs/fine_logs/clean_services_logs_action.py

@@ -0,0 +1,83 @@
+# docs: docs/reference/actions.md
+import dataclasses
+
+from finecode_extension_api import code_action, textstyler
+
+
+@dataclasses.dataclass
+class CleanServicesLogsRunPayload(code_action.RunActionPayload):
+    service_ids: list[str] | None = None
+    """Services whose logs should be deleted.
+    None = discover and clean all available services (requires a discovery handler).
+    Empty list = explicit no-op."""
+
+
+class CleanServicesLogsRunContext(
+    code_action.RunActionContext[CleanServicesLogsRunPayload]
+):
+    def __init__(
+        self,
+        run_id: int,
+        initial_payload: CleanServicesLogsRunPayload,
+        meta: code_action.RunActionMeta,
+        info_provider: code_action.RunContextInfoProvider,
+        progress_sender: code_action.ProgressSender = code_action._NOOP_PROGRESS_SENDER,
+    ) -> None:
+        super().__init__(
+            run_id=run_id,
+            initial_payload=initial_payload,
+            meta=meta,
+            info_provider=info_provider,
+            progress_sender=progress_sender,
+        )
+        self.service_ids: list[str] | None = None
+        """None = discovery has not run. [] = explicit no-op."""
+
+    async def init(self) -> None:
+        if self.initial_payload.service_ids is not None:
+            self.service_ids = list(self.initial_payload.service_ids)
+
+
+@dataclasses.dataclass
+class CleanServicesLogsRunResult(code_action.RunActionResult):
+    services_cleaned: list[str] = dataclasses.field(default_factory=list)
+    errors: list[str] = dataclasses.field(default_factory=list)
+
+    def update(self, other: code_action.RunActionResult) -> None:
+        if not isinstance(other, CleanServicesLogsRunResult):
+            return
+        self.services_cleaned += other.services_cleaned
+        self.errors += other.errors
+
+    def to_text(self) -> str | textstyler.StyledText:
+        if self.errors:
+            return "\n".join(self.errors)
+        return f"Cleaned {len(self.services_cleaned)} service(s)."
+
+    @property
+    def return_code(self) -> code_action.RunReturnCode:
+        return (
+            code_action.RunReturnCode.ERROR
+            if self.errors
+            else code_action.RunReturnCode.SUCCESS
+        )
+
+
+class CleanServicesLogsAction(
+    code_action.Action[
+        CleanServicesLogsRunPayload,
+        CleanServicesLogsRunContext,
+        CleanServicesLogsRunResult,
+    ]
+):
+    """Delete logs for multiple observability services.
+
+    When service_ids is None, a discovery handler must run first to populate
+    run_context.service_ids from list_observability_services. No-op when
+    service_ids is an empty list.
+    """
+
+    DESCRIPTION = "Delete logs for multiple observability services."
+    PAYLOAD_TYPE = CleanServicesLogsRunPayload
+    RUN_CONTEXT_TYPE = CleanServicesLogsRunContext
+    RESULT_TYPE = CleanServicesLogsRunResult

presets/fine_logs/fine_logs/clean_services_logs_discovery_handler.py

@@ -0,0 +1,56 @@
+import dataclasses
+
+from finecode_extension_api import code_action
+from fine_logs.clean_services_logs_action import (
+    CleanServicesLogsAction,
+    CleanServicesLogsRunContext,
+    CleanServicesLogsRunPayload,
+    CleanServicesLogsRunResult,
+)
+from fine_logs.list_observability_services_action import (
+    ListObservabilityServicesAction,
+    ListObservabilityServicesRunPayload,
+)
+from finecode_extension_api.interfaces import ilogger, iprojectactionrunner
+
+
+@dataclasses.dataclass
+class CleanServicesLogsDiscoveryHandlerConfig(code_action.ActionHandlerConfig): ...
+
+
+class CleanServicesLogsDiscoveryHandler(
+    code_action.ActionHandler[
+        CleanServicesLogsAction,
+        CleanServicesLogsDiscoveryHandlerConfig,
+    ]
+):
+    """Populate run_context.service_ids by calling list_observability_services.
+
+    Must be registered before performing cleaning. No-op when
+    run_context.service_ids is already set (i.e. explicit caller or repeated run).
+    """
+
+    def __init__(
+        self,
+        project_action_runner: iprojectactionrunner.IProjectActionRunner,
+        logger: ilogger.ILogger,
+    ) -> None:
+        self.project_action_runner = project_action_runner
+        self.logger = logger
+
+    async def run(
+        self,
+        payload: CleanServicesLogsRunPayload,
+        run_context: CleanServicesLogsRunContext,
+    ) -> CleanServicesLogsRunResult:
+        if run_context.service_ids is not None:
+            return CleanServicesLogsRunResult()
+
+        result = await self.project_action_runner.run_action(
+            action_type=iprojectactionrunner.ActionRef.from_type(ListObservabilityServicesAction),
+            payload=ListObservabilityServicesRunPayload(),
+            meta=run_context.meta,
+        )
+        run_context.service_ids = [s.service_id for s in result.services]
+        self.logger.debug(f"Discovered services to clean: {run_context.service_ids}")
+        return CleanServicesLogsRunResult()

presets/fine_logs/fine_logs/clean_services_logs_iterate_handler.py

@@ -0,0 +1,70 @@
+import dataclasses
+
+from finecode_extension_api import code_action
+from fine_logs.clean_service_logs_action import (
+    CleanServiceLogsAction,
+    CleanServiceLogsRunPayload,
+)
+from fine_logs.clean_services_logs_action import (
+    CleanServicesLogsAction,
+    CleanServicesLogsRunContext,
+    CleanServicesLogsRunPayload,
+    CleanServicesLogsRunResult,
+)
+from finecode_extension_api.interfaces import ilogger, iprojectactionrunner
+
+
+@dataclasses.dataclass
+class CleanServicesLogsIterateHandlerConfig(code_action.ActionHandlerConfig): ...
+
+
+class CleanServicesLogsIterateHandler(
+    code_action.ActionHandler[
+        CleanServicesLogsAction,
+        CleanServicesLogsIterateHandlerConfig,
+    ]
+):
+    """Call clean_service_logs for each service_id in run_context.service_ids.
+
+    Must be registered after discovery handler.
+    """
+
+    def __init__(
+        self,
+        project_action_runner: iprojectactionrunner.IProjectActionRunner,
+        logger: ilogger.ILogger,
+    ) -> None:
+        self.project_action_runner = project_action_runner
+        self.logger = logger
+
+    async def run(
+        self,
+        payload: CleanServicesLogsRunPayload,
+        run_context: CleanServicesLogsRunContext,
+    ) -> CleanServicesLogsRunResult:
+        if run_context.service_ids is None:
+            raise code_action.ActionFailedException(
+                "CleanServicesLogsDiscoveryHandler must be registered before"
+                " CleanServicesLogsIterateHandler"
+            )
+        if not run_context.service_ids:
+            return CleanServicesLogsRunResult()
+
+        services_cleaned: list[str] = []
+        errors: list[str] = []
+
+        for service_id in run_context.service_ids:
+            result = await self.project_action_runner.run_action(
+                action_type=iprojectactionrunner.ActionRef.from_type(CleanServiceLogsAction),
+                payload=CleanServiceLogsRunPayload(service_id=service_id),
+                meta=run_context.meta,
+            )
+            if result.errors:
+                errors.extend(result.errors)
+            else:
+                services_cleaned.append(service_id)
+
+        return CleanServicesLogsRunResult(
+            services_cleaned=services_cleaned,
+            errors=errors,
+        )