"""Process Monitor — Resource worker layer.""" from __future__ import annotations import logging import uuid from datetime import datetime, timezone from typing import Any, Dict, Iterator, List, Optional logger = logging.getLogger(__name__) class ProcessWorker: """Resource worker for the Process Monitor application.""" def __init__( self, store: Any, config: Optional[Dict[str, Any]] = None, ) -> None: self._store = store self._cfg = config or {} self._status = self._cfg.get("status", None) logger.debug("%s initialised", self.__class__.__name__) def watch_resource( self, status: Any, cpu_pct: Any, **extra: Any ) -> Dict[str, Any]: """Create and persist a new Resource record.""" now = datetime.now(timezone.utc).isoformat() record: Dict[str, Any] = { "id": str(uuid.uuid4()), "status": status, "cpu_pct": cpu_pct, "status": "active", "created_at": now, **extra, } saved = self._store.put(record) logger.info("watch_resource: created %s", saved["id"]) return saved def get_resource(self, record_id: str) -> Optional[Dict[str, Any]]: """Retrieve a Resource by its *record_id*.""" record = self._store.get(record_id) if record is None: logger.debug("get_resource: %s not found", record_id) return record def snapshot_resource( self, record_id: str, **changes: Any ) -> Dict[str, Any]: """Apply *changes* to an existing Resource.""" record = self._store.get(record_id) if record is None: raise KeyError(f"Resource {record_id!r} not found") record.update(changes) record["updated_at"] = datetime.now(timezone.utc).isoformat() return self._store.put(record) def kill_resource(self, record_id: str) -> bool: """Remove a Resource; returns True on success.""" if self._store.get(record_id) is None: return False self._store.delete(record_id) logger.info("kill_resource: removed %s", record_id) return True def list_resources( self, status: Optional[str] = None, limit: int = 50, offset: int = 0, ) -> List[Dict[str, Any]]: """Return paginated Resource records.""" query: Dict[str, Any] = {} if status: query["status"] = status results = self._store.find(query, limit=limit, offset=offset) logger.debug("list_resources: %d results", len(results)) return results def iter_resources( self, batch_size: int = 100 ) -> Iterator[Dict[str, Any]]: """Yield all Resource records in batches of *batch_size*.""" offset = 0 while True: page = self.list_resources(limit=batch_size, offset=offset) if not page: break yield from page if len(page) < batch_size: break offset += batch_size