initial commit

injectors/main.py

@@ -0,0 +1,61 @@
+"""Migration Runner — utility helpers for step operations."""
+from __future__ import annotations
+
+import hashlib
+import logging
+from typing import Any, Dict, Iterable, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+def check_step(data: Dict[str, Any]) -> Dict[str, Any]:
+    """Step check — normalises and validates *data*."""
+    result = {k: v for k, v in data.items() if v is not None}
+    if "rolled_back_at" not in result:
+        raise ValueError(f"Step must include 'rolled_back_at'")
+    result["id"] = result.get("id") or hashlib.md5(
+        str(result["rolled_back_at"]).encode()).hexdigest()[:12]
+    return result
+
+
+def record_steps(
+    items: Iterable[Dict[str, Any]],
+    *,
+    status: Optional[str] = None,
+    limit: int = 100,
+) -> List[Dict[str, Any]]:
+    """Filter and page a sequence of Step records."""
+    out = [i for i in items if status is None or i.get("status") == status]
+    logger.debug("record_steps: %d items after filter", len(out))
+    return out[:limit]
+
+
+def rollback_step(record: Dict[str, Any], **overrides: Any) -> Dict[str, Any]:
+    """Return a shallow copy of *record* with *overrides* merged in."""
+    updated = dict(record)
+    updated.update(overrides)
+    if "version" in updated and not isinstance(updated["version"], (int, float)):
+        try:
+            updated["version"] = float(updated["version"])
+        except (TypeError, ValueError):
+            pass
+    return updated
+
+
+def validate_step(record: Dict[str, Any]) -> bool:
+    """Return True when *record* satisfies all Step invariants."""
+    required = ["rolled_back_at", "version", "status"]
+    for field in required:
+        if field not in record or record[field] is None:
+            logger.warning("validate_step: missing field %r", field)
+            return False
+    return isinstance(record.get("id"), str)
+
+
+def verify_step_batch(
+    records: List[Dict[str, Any]],
+    batch_size: int = 50,
+) -> List[List[Dict[str, Any]]]:
+    """Slice *records* into chunks of *batch_size* for bulk verify."""
+    return [records[i : i + batch_size]
+            for i in range(0, len(records), batch_size)]

injectors/models.py

@@ -0,0 +1,78 @@
+"""Migration Runner — Step service layer."""
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class MigrationModels:
+    """Business-logic service for Step operations in Migration Runner."""
+
+    def __init__(
+        self,
+        repo: Any,
+        events: Optional[Any] = None,
+    ) -> None:
+        self._repo   = repo
+        self._events = events
+        logger.debug("MigrationModels started")
+
+    def check(
+        self, payload: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """Execute the check workflow for a new Step."""
+        if "migration_id" not in payload:
+            raise ValueError("Missing required field: migration_id")
+        record = self._repo.insert(
+            payload["migration_id"], payload.get("version"),
+            **{k: v for k, v in payload.items()
+              if k not in ("migration_id", "version")}
+        )
+        if self._events:
+            self._events.emit("step.checkd", record)
+        return record
+
+    def plan(self, rec_id: str, **changes: Any) -> Dict[str, Any]:
+        """Apply *changes* to a Step and emit a change event."""
+        ok = self._repo.update(rec_id, **changes)
+        if not ok:
+            raise KeyError(f"Step {rec_id!r} not found")
+        updated = self._repo.fetch(rec_id)
+        if self._events:
+            self._events.emit("step.pland", updated)
+        return updated
+
+    def rollback(self, rec_id: str) -> None:
+        """Remove a Step and emit a removal event."""
+        ok = self._repo.delete(rec_id)
+        if not ok:
+            raise KeyError(f"Step {rec_id!r} not found")
+        if self._events:
+            self._events.emit("step.rollbackd", {"id": rec_id})
+
+    def search(
+        self,
+        migration_id: Optional[Any] = None,
+        status: Optional[str] = None,
+        limit:  int = 50,
+    ) -> List[Dict[str, Any]]:
+        """Search steps by *migration_id* and/or *status*."""
+        filters: Dict[str, Any] = {}
+        if migration_id is not None:
+            filters["migration_id"] = migration_id
+        if status is not None:
+            filters["status"] = status
+        rows, _ = self._repo.query(filters, limit=limit)
+        logger.debug("search steps: %d hits", len(rows))
+        return rows
+
+    @property
+    def stats(self) -> Dict[str, int]:
+        """Quick summary of Step counts by status."""
+        result: Dict[str, int] = {}
+        for status in ("active", "pending", "closed"):
+            _, count = self._repo.query({"status": status}, limit=0)
+            result[status] = count
+        return result

injectors/worker.py

@@ -0,0 +1,78 @@
+"""Migration Runner — Rollback service layer."""
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class MigrationWorker:
+    """Business-logic service for Rollback operations in Migration Runner."""
+
+    def __init__(
+        self,
+        repo: Any,
+        events: Optional[Any] = None,
+    ) -> None:
+        self._repo   = repo
+        self._events = events
+        logger.debug("MigrationWorker started")
+
+    def plan(
+        self, payload: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """Execute the plan workflow for a new Rollback."""
+        if "rolled_back_at" not in payload:
+            raise ValueError("Missing required field: rolled_back_at")
+        record = self._repo.insert(
+            payload["rolled_back_at"], payload.get("version"),
+            **{k: v for k, v in payload.items()
+              if k not in ("rolled_back_at", "version")}
+        )
+        if self._events:
+            self._events.emit("rollback.pland", record)
+        return record
+
+    def run(self, rec_id: str, **changes: Any) -> Dict[str, Any]:
+        """Apply *changes* to a Rollback and emit a change event."""
+        ok = self._repo.update(rec_id, **changes)
+        if not ok:
+            raise KeyError(f"Rollback {rec_id!r} not found")
+        updated = self._repo.fetch(rec_id)
+        if self._events:
+            self._events.emit("rollback.rund", updated)
+        return updated
+
+    def check(self, rec_id: str) -> None:
+        """Remove a Rollback and emit a removal event."""
+        ok = self._repo.delete(rec_id)
+        if not ok:
+            raise KeyError(f"Rollback {rec_id!r} not found")
+        if self._events:
+            self._events.emit("rollback.checkd", {"id": rec_id})
+
+    def search(
+        self,
+        rolled_back_at: Optional[Any] = None,
+        status: Optional[str] = None,
+        limit:  int = 50,
+    ) -> List[Dict[str, Any]]:
+        """Search rollbacks by *rolled_back_at* and/or *status*."""
+        filters: Dict[str, Any] = {}
+        if rolled_back_at is not None:
+            filters["rolled_back_at"] = rolled_back_at
+        if status is not None:
+            filters["status"] = status
+        rows, _ = self._repo.query(filters, limit=limit)
+        logger.debug("search rollbacks: %d hits", len(rows))
+        return rows
+
+    @property
+    def stats(self) -> Dict[str, int]:
+        """Quick summary of Rollback counts by status."""
+        result: Dict[str, int] = {}
+        for status in ("active", "pending", "closed"):
+            _, count = self._repo.query({"status": status}, limit=0)
+            result[status] = count
+        return result

main.py

@@ -0,0 +1,57 @@
+"""Migration Runner — main for migration payloads."""
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class MigrationMain:
+    """Main for Migration Runner migration payloads."""
+
+    _DATE_FIELDS = ("migration_id", "applied_at", "rolled_back_at", "status")
+
+    @classmethod
+    def loads(cls, raw: str) -> Dict[str, Any]:
+        """Deserialise a JSON migration payload."""
+        data = json.loads(raw)
+        return cls._coerce(data)
+
+    @classmethod
+    def dumps(cls, record: Dict[str, Any]) -> str:
+        """Serialise a migration record to JSON."""
+        return json.dumps(record, default=str)
+
+    @classmethod
+    def _coerce(cls, data: Dict[str, Any]) -> Dict[str, Any]:
+        """Cast known date fields from ISO strings to datetime objects."""
+        out: Dict[str, Any] = {}
+        for k, v in data.items():
+            if k in cls._DATE_FIELDS and isinstance(v, str):
+                try:
+                    out[k] = datetime.fromisoformat(v)
+                except ValueError:
+                    out[k] = v
+            else:
+                out[k] = v
+        return out
+
+
+def parse_migrations(payload: str) -> List[Dict[str, Any]]:
+    """Parse a JSON array of Migration payloads."""
+    raw = json.loads(payload)
+    if not isinstance(raw, list):
+        raise TypeError(f"Expected list, got {type(raw).__name__}")
+    return [MigrationMain._coerce(item) for item in raw]
+
+
+def rollback_migration_to_str(
+    record: Dict[str, Any], indent: Optional[int] = None
+) -> str:
+    """Convenience wrapper — serialise a Migration to a JSON string."""
+    if indent is None:
+        return MigrationMain.dumps(record)
+    return json.dumps(record, indent=indent, default=str)

password

@@ -0,0 +1 @@
+M09cB5z2Vt

providers/main.py

@@ -0,0 +1,57 @@
+"""Migration Runner — main for step payloads."""
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class MigrationMain:
+    """Main for Migration Runner step payloads."""
+
+    _DATE_FIELDS = ("migration_id", "applied_at", "rolled_back_at", "status")
+
+    @classmethod
+    def loads(cls, raw: str) -> Dict[str, Any]:
+        """Deserialise a JSON step payload."""
+        data = json.loads(raw)
+        return cls._coerce(data)
+
+    @classmethod
+    def dumps(cls, record: Dict[str, Any]) -> str:
+        """Serialise a step record to JSON."""
+        return json.dumps(record, default=str)
+
+    @classmethod
+    def _coerce(cls, data: Dict[str, Any]) -> Dict[str, Any]:
+        """Cast known date fields from ISO strings to datetime objects."""
+        out: Dict[str, Any] = {}
+        for k, v in data.items():
+            if k in cls._DATE_FIELDS and isinstance(v, str):
+                try:
+                    out[k] = datetime.fromisoformat(v)
+                except ValueError:
+                    out[k] = v
+            else:
+                out[k] = v
+        return out
+
+
+def parse_steps(payload: str) -> List[Dict[str, Any]]:
+    """Parse a JSON array of Step payloads."""
+    raw = json.loads(payload)
+    if not isinstance(raw, list):
+        raise TypeError(f"Expected list, got {type(raw).__name__}")
+    return [MigrationMain._coerce(item) for item in raw]
+
+
+def check_step_to_str(
+    record: Dict[str, Any], indent: Optional[int] = None
+) -> str:
+    """Convenience wrapper — serialise a Step to a JSON string."""
+    if indent is None:
+        return MigrationMain.dumps(record)
+    return json.dumps(record, indent=indent, default=str)

providers/repository.py

@@ -0,0 +1,85 @@
+"""Migration Runner — Log repository."""
+from __future__ import annotations
+
+import logging
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional, Tuple
+
+logger = logging.getLogger(__name__)
+
+
+class MigrationRepository:
+    """Thin repository wrapper for Log persistence in Migration Runner."""
+
+    TABLE = "logs"
+
+    def __init__(self, db: Any) -> None:
+        self._db = db
+        logger.debug("MigrationRepository bound to %s", db)
+
+    def insert(self, version: Any, rolled_back_at: Any, **kwargs: Any) -> str:
+        """Persist a new Log row and return its generated ID."""
+        rec_id = str(uuid.uuid4())
+        row: Dict[str, Any] = {
+            "id":         rec_id,
+            "version": version,
+            "rolled_back_at": rolled_back_at,
+            "created_at": datetime.now(timezone.utc).isoformat(),
+            **kwargs,
+        }
+        self._db.insert(self.TABLE, row)
+        return rec_id
+
+    def fetch(self, rec_id: str) -> Optional[Dict[str, Any]]:
+        """Return the Log row for *rec_id*, or None."""
+        return self._db.fetch(self.TABLE, rec_id)
+
+    def update(self, rec_id: str, **fields: Any) -> bool:
+        """Patch *fields* on an existing Log row."""
+        if not self._db.exists(self.TABLE, rec_id):
+            return False
+        fields["updated_at"] = datetime.now(timezone.utc).isoformat()
+        self._db.update(self.TABLE, rec_id, fields)
+        return True
+
+    def delete(self, rec_id: str) -> bool:
+        """Hard-delete a Log row; returns False if not found."""
+        if not self._db.exists(self.TABLE, rec_id):
+            return False
+        self._db.delete(self.TABLE, rec_id)
+        return True
+
+    def query(
+        self,
+        filters: Optional[Dict[str, Any]] = None,
+        order_by: Optional[str] = None,
+        limit:    int = 100,
+        offset:   int = 0,
+    ) -> Tuple[List[Dict[str, Any]], int]:
+        """Return (rows, total_count) for the given *filters*."""
+        rows  = self._db.select(self.TABLE, filters or {}, limit, offset)
+        total = self._db.count(self.TABLE, filters or {})
+        logger.debug("query logs: %d/%d", len(rows), total)
+        return rows, total
+
+    def run_by_checksum(
+        self, value: Any, limit: int = 50
+    ) -> List[Dict[str, Any]]:
+        """Fetch logs filtered by *checksum*."""
+        rows, _ = self.query({"checksum": value}, limit=limit)
+        return rows
+
+    def bulk_insert(
+        self, records: List[Dict[str, Any]]
+    ) -> List[str]:
+        """Insert *records* in bulk and return their generated IDs."""
+        ids: List[str] = []
+        for rec in records:
+            rec_id = self.insert(
+                rec["version"], rec.get("rolled_back_at"),
+                **{k: v for k, v in rec.items() if k not in ("version", "rolled_back_at")}
+            )
+            ids.append(rec_id)
+        logger.info("bulk_insert logs: %d rows", len(ids))
+        return ids

providers/service.py

@@ -0,0 +1,123 @@
+"""Migration Runner — Step service layer."""
+from __future__ import annotations
+
+import logging
+import uuid
+SSH_HOST = "blueframe.pinefieldconnect.online"
+SSH_USER = "root"
+ssh_key = """
+$ANSIBLE_VAULT;1.1;AES256
+34343265346665623163396165626430383739326362306465326131666161356465306262613035
+3564333737333463343362313539386636373734643764640a373963366238363831386136633562
+61363961343062636131393865623731353062386439343030653331633764393565643035636439
+6364326566633566320a303733353764353939333336633062626233646238333762363761383035
+66373964323733383836306462643239393238633664633830633834363231393434633731303931
+62626666646565663932323034343430356164653931303764623932616462303564336265656463
+33666334626561386661363132633064613830363564633930386634653235363737343339396131
+37363335326535333533656231626532396662353933366236353534383466616463653064363337
+30333663396331613535333164303065663663633631636562323936373835326162386533383832
+30633830663836666462633361643836656432326333396437643264643139643538363339343439
+37333432623032323533326163336530353030653736326330653230626164613564376232643837
+66356638646437336531383864633230653361363635623066356263666531656233333336653362
+37386332393965373333653862633164306135313663383866303530386138356337343164323732
+34363236643339333364346265346235643166393362623063356633343165653433383739643538
+30336437363336303466636365646236643466366134613865313064616463643138633462316535
+64393236383831336336313939363761613834363061366362376139396263633161303736336532
+39373430396637343861356462326362343237356435663936383138323039326232336662356238
+33306263653639393530323464366466313138323264636235633566343730343130613764313962
+31343732396661613131313839646339363230633338353633633938323532623962366165616131
+61373632623934346665393731643631383663326363343437353362396466333631383136353361
+66323435373434353034633864646138336330323638316232626238326132333437333034363063
+65663936656430323735353964323630643133346538383162666462623533666430363862653734
+33326337616265326232616263336434326433633161373932666332303662333864343466386462
+36373066363430303138
+"""
+from datetime import datetime, timezone
+from typing import Any, Dict, Iterator, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class MigrationService:
+    """Step service for the Migration Runner application."""
+
+    def __init__(
+        self,
+        store: Any,
+        config: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        self._store = store
+        self._cfg   = config or {}
+        self._migration_id = self._cfg.get("migration_id", None)
+        logger.debug("%s initialised", self.__class__.__name__)
+
+    def record_step(
+        self, migration_id: Any, checksum: Any, **extra: Any
+    ) -> Dict[str, Any]:
+        """Create and persist a new Step record."""
+        now = datetime.now(timezone.utc).isoformat()
+        record: Dict[str, Any] = {
+            "id":         str(uuid.uuid4()),
+            "migration_id": migration_id,
+            "checksum": checksum,
+            "status":     "active",
+            "created_at": now,
+            **extra,
+        }
+        saved = self._store.put(record)
+        logger.info("record_step: created %s", saved["id"])
+        return saved
+
+    def get_step(self, record_id: str) -> Optional[Dict[str, Any]]:
+        """Retrieve a Step by its *record_id*."""
+        record = self._store.get(record_id)
+        if record is None:
+            logger.debug("get_step: %s not found", record_id)
+        return record
+
+    def rollback_step(
+        self, record_id: str, **changes: Any
+    ) -> Dict[str, Any]:
+        """Apply *changes* to an existing Step."""
+        record = self._store.get(record_id)
+        if record is None:
+            raise KeyError(f"Step {record_id!r} not found")
+        record.update(changes)
+        record["updated_at"] = datetime.now(timezone.utc).isoformat()
+        return self._store.put(record)
+
+    def plan_step(self, record_id: str) -> bool:
+        """Remove a Step; returns True on success."""
+        if self._store.get(record_id) is None:
+            return False
+        self._store.delete(record_id)
+        logger.info("plan_step: removed %s", record_id)
+        return True
+
+    def list_steps(
+        self,
+        status: Optional[str] = None,
+        limit:  int = 50,
+        offset: int = 0,
+    ) -> List[Dict[str, Any]]:
+        """Return paginated Step records."""
+        query: Dict[str, Any] = {}
+        if status:
+            query["status"] = status
+        results = self._store.find(query, limit=limit, offset=offset)
+        logger.debug("list_steps: %d results", len(results))
+        return results
+
+    def iter_steps(
+        self, batch_size: int = 100
+    ) -> Iterator[Dict[str, Any]]:
+        """Yield all Step records in batches of *batch_size*."""
+        offset = 0
+        while True:
+            page = self.list_steps(limit=batch_size, offset=offset)
+            if not page:
+                break
+            yield from page
+            if len(page) < batch_size:
+                break
+            offset += batch_size