devplacepy/store.py at db56ebbe74c51d59441fec1a33de37e1fef2d04b

 # retoor <retoor@molodetz.nl>
 import json
 from datetime import datetime, timezone
 from devplacepy.database import db, get_table
 from devplacepy.utils import generate_uid, make_combined_slug
 ST_CREATED = "created"
 ST_STARTING = "starting"
 ST_RUNNING = "running"
 ST_STOPPED = "stopped"
 ST_PAUSED = "paused"
 ST_CRASHED = "crashed"
 ST_RESTARTING = "restarting"
 ST_REMOVING = "removing"
 ST_REMOVED = "removed"
 DESIRED_RUNNING = "running"
 DESIRED_STOPPED = "stopped"
 DESIRED_PAUSED = "paused"
 RESTART_POLICIES = ("never", "always", "on-failure", "unless-stopped")
 METRICS_RING = 720
 def now() -> str:
     return datetime.now(timezone.utc).isoformat()
 def _exists(table: str) -> bool:
     return table in db.tables
 # ---------------- instances ----------------
 def create_instance(row: dict) -> dict:
     uid = generate_uid()
     base = {
         "uid": uid,
         "slug": make_combined_slug(row.get("name", "instance"), uid),
         "container_id": "",
         "status": ST_CREATED,
         "exit_code": 0,
         "restart_count": 0,
         "ingress_slug": "",
         "ingress_port": 0,
         "started_at": "",
         "stopped_at": "",
         "created_at": now(),
         "updated_at": now(),
     }
     base.update(row)
     base["uid"] = uid
     get_table("instances").insert(base)
     return get_instance(uid)
 def get_instance(uid: str):
     table = get_table("instances")
     if not _exists("instances"):
         return None
     return table.find_one(uid=uid) or table.find_one(slug=uid)
 def list_instances(project_uid: str = None) -> list:
     if not _exists("instances"):
         return []
     filters = {}
     if project_uid:
         filters["project_uid"] = project_uid
     return sorted(
         get_table("instances").find(**filters),
         key=lambda r: r.get("created_at", ""),
         reverse=True,
     )
 def all_instances() -> list:
     if not _exists("instances"):
         return []
     return list(get_table("instances").find())
 def find_instance_by_ingress(slug: str):
     if not slug or not _exists("instances"):
         return None
     matches = list(get_table("instances").find(ingress_slug=slug))
     for instance in matches:
         if instance.get("status") == ST_RUNNING:
             return instance
     return matches[0] if matches else None
 def update_instance(uid: str, changes: dict) -> None:
     get_table("instances").update({"uid": uid, "updated_at": now(), **changes}, ["uid"])
 def delete_instance(uid: str) -> None:
     get_table("instances").delete(uid=uid)
 # ---------------- events / metrics / schedules ----------------
 def record_event(
     instance: dict, event: str, actor_kind: str, actor_id: str, detail: dict = None
 ) -> None:
     get_table("instance_events").insert(
         {
             "uid": generate_uid(),
             "instance_uid": instance["uid"],
             "project_uid": instance.get("project_uid", ""),
             "event": event,
             "actor_kind": actor_kind,
             "actor_id": actor_id or "",
             "detail": json.dumps(detail or {}),
             "created_at": now(),
         }
     )
 def list_events(instance_uid: str, limit: int = 100) -> list:
     if not _exists("instance_events"):
         return []
     rows = sorted(
         get_table("instance_events").find(instance_uid=instance_uid),
         key=lambda r: r.get("created_at", ""),
         reverse=True,
     )
     return rows[:limit]
 def insert_metric(instance_uid: str, sample) -> None:
     table = get_table("instance_metrics")
     table.insert(
         {
             "uid": generate_uid(),
             "instance_uid": instance_uid,
             "ts": now(),
             "cpu_pct": sample.cpu_pct,
             "mem_bytes": sample.mem_bytes,
             "mem_pct": sample.mem_pct,
             "net_rx": sample.net_rx,
             "net_tx": sample.net_tx,
             "blk_read": sample.blk_read,
             "blk_write": sample.blk_write,
         }
     )
     rows = sorted(table.find(instance_uid=instance_uid), key=lambda r: r.get("ts", ""))
     if len(rows) > METRICS_RING:
         for old in rows[: len(rows) - METRICS_RING]:
             table.delete(uid=old["uid"])
 def recent_metrics(instance_uid: str, limit: int = 120) -> list:
     if not _exists("instance_metrics"):
         return []
     rows = sorted(
         get_table("instance_metrics").find(instance_uid=instance_uid),
         key=lambda r: r.get("ts", ""),
     )
     return rows[-limit:]
 def create_schedule(
     instance: dict, action: str, schedule_columns: dict, next_run_at: str
 ) -> dict:
     uid = generate_uid()
     get_table("instance_schedules").insert(
         {
             "uid": uid,
             "instance_uid": instance["uid"],
             "project_uid": instance.get("project_uid", ""),
             "action": action,
             "schedule_json": json.dumps(schedule_columns),
             "enabled": 1,
             "next_run_at": next_run_at,
             "last_run_at": "",
             "run_count": 0,
             "created_at": now(),
             "updated_at": now(),
         }
     )
     return get_table("instance_schedules").find_one(uid=uid)
 def list_schedules(instance_uid: str) -> list:
     if not _exists("instance_schedules"):
         return []
     return list(get_table("instance_schedules").find(instance_uid=instance_uid))
 def due_schedules(now_iso: str) -> list:
     if not _exists("instance_schedules"):
         return []
     return [
         r
         for r in get_table("instance_schedules").find(enabled=1)
         if r.get("next_run_at") and r["next_run_at"] <= now_iso
     ]
 def update_schedule(uid: str, changes: dict) -> None:
     get_table("instance_schedules").update(
         {"uid": uid, "updated_at": now(), **changes}, ["uid"]
     )
 def delete_schedule(uid: str) -> None:
     get_table("instance_schedules").delete(uid=uid)