Merge pull request #21 from Sreejay-Reddy/feature/CLI-inspect

feat: add inspect API, sentinel CLI, and decouple reconciliation from once()

Author: Sreejay-Reddy

sentinel-py/pyproject.toml

@@ -35,8 +35,12 @@ classifiers = [
 
 license = { file = "LICENSE" }
 
+[project.scripts]
+sen = "sentinel.cli:main"
+
 [project.optional-dependencies]
 django = ["django>=4.2"]
+cli = ["python-dotenv"]
 
 [build-system]
 requires = ["setuptools>=61.0"]

sentinel-py/sentinel/async_core.py

@@ -1,6 +1,6 @@
 import json
 from .utils import get_owner_id, row_to_dict
-from .result import AcquireResult, OperationResult
+from .result import AcquireResult, OperationResult, InspectResult
 
 async def acquire(conn, key, *, owner_id=None, ttl_ms=10000, hard_ttl_ms = None):
 
@@ -182,4 +182,30 @@ async def expire_lease(conn, key, *, owner_id, fencing_token):
     await conn.commit()
     return OperationResult(success)
 
+async def inspect(conn, key):
+    async with conn.cursor() as cur:
+        await cur.execute("""
+        SELECT owner_id, lease_expires_at, fencing_token, status, 
+        lease_expires_at > NOW() AS lease_alive, lease_updated_at,
+        hard_expires_at, execution_result
+        FROM sentinel_leases
+        WHERE key = %s 
+        """, (key,))
+
+        result = await cur.fetchone()
 
+        if result is None:
+            return None
+
+        row = row_to_dict(cur, result)
+
+    return InspectResult(
+        key = key,
+        owner_id = row["owner_id"],
+        fencing_token = row["fencing_token"],
+        status = row["status"],
+        lease_alive = row["lease_alive"],
+        lease_expires_at = row["lease_expires_at"],
+        lease_updated_at = row["lease_updated_at"],
+        hard_expires_at = row["hard_expires_at"],
+        execution_result = row["execution_result"])

sentinel-py/sentinel/async_once.py

@@ -13,7 +13,6 @@
 
 from .heartbeat_config import get_manager
 from .logging import logger
-from .async_reconcilliation import AsyncReconcile
 from .result import OnceResult
 
 
@@ -36,7 +35,6 @@ def __init__(
         self.kwargs = kwargs or {}
         self.owns_connection = owns_connection
 
-        self.reconcile = AsyncReconcile(get_conn)
         self._task = None
 
     async def run(self):
@@ -76,8 +74,7 @@ async def run(self):
                         success=False,
                         status="executing",
                         uncertain=True,
-                        execution_alive=False,
-                        reconcile=self.reconcile
+                        execution_alive=False
                     )
 
                 if acquired.status == "executing" and acquired.lease_alive:
@@ -162,8 +159,7 @@ async def run(self):
                     status="executing",
                     execution_alive=False,
                     uncertain=True,
-                    exception=e,
-                    reconcile=self.reconcile
+                    exception=e
                 )
 
             # Finalize canonical completion
@@ -184,8 +180,7 @@ async def run(self):
                 return OnceResult(
                     success=False,
                     status=completed.status,
-                    uncertain=True,
-                    reconcile=self.reconcile
+                    uncertain=True
                 )
 
             return OnceResult(

sentinel-py/sentinel/async_reconcilliation.py

@@ -6,7 +6,7 @@ def __init__(self, get_conn, namespace=None):
         self.get_conn = get_conn
         self.namespace = namespace
 
-    async def reconcile(self, key, *, owner_id, fencing_token):
+    async def reconcile(self, key):
         conn = await self.get_conn()
 
         try:
@@ -17,12 +17,10 @@ async def reconcile(self, key, *, owner_id, fencing_token):
                     status = 'reconciling',
                     lease_updated_at = NOW()
                 WHERE key = %s
-                  AND owner_id = %s
-                  AND fencing_token = %s
                   AND status = 'executing'
                   AND lease_expires_at < NOW()
                 RETURNING 1;
-                """, (key, owner_id, fencing_token))
+                """, (key,))
 
                 success = await cur.fetchone() is not None
 
@@ -33,14 +31,7 @@ async def reconcile(self, key, *, owner_id, fencing_token):
         finally:
             await conn.close()
 
-    async def force_complete(
-        self,
-        key,
-        *,
-        owner_id,
-        fencing_token,
-        execution_result
-    ):
+    async def force_complete(self, key, execution_result):
         conn = await self.get_conn()
 
         try:
@@ -52,16 +43,9 @@ async def force_complete(
                     execution_result = %s,
                     lease_updated_at = NOW()
                 WHERE key = %s
-                  AND owner_id = %s
-                  AND fencing_token = %s
                   AND status = 'reconciling'
                 RETURNING 1;
-                """, (
-                    execution_result,
-                    key,
-                    owner_id,
-                    fencing_token
-                ))
+                """, (execution_result, key))
 
                 success = await cur.fetchone() is not None
 
@@ -72,13 +56,7 @@ async def force_complete(
         finally:
             await conn.close()
 
-    async def reset(
-        self,
-        key,
-        *,
-        owner_id,
-        fencing_token
-    ):
+    async def reset(self, key):
         conn = await self.get_conn()
 
         try:
@@ -89,15 +67,9 @@ async def reset(
                     status = 'claimed',
                     lease_updated_at = NOW()
                 WHERE key = %s
-                  AND owner_id = %s
-                  AND fencing_token = %s
                   AND status = 'reconciling'
                 RETURNING 1;
-                """, (
-                    key,
-                    owner_id,
-                    fencing_token
-                ))
+                """, (key,))
 
                 success = await cur.fetchone() is not None
 

sentinel-py/sentinel/async_sentinel.py

@@ -1,4 +1,4 @@
-from .lease import Lease
+from .async_reconcilliation import AsyncReconcile
 from .async_once import AsyncOnce
 from .heartbeat_config import get_manager
 
@@ -16,6 +16,8 @@ def __init__(self, get_conn = None, default_ttl_ms=3000, namespace=None, owns_co
             raise ValueError(
                 "No database connection provider found."
             )
+        
+        self.reconcile = AsyncReconcile(self._conn, namespace=self.namespace)
 
         self.manager = get_manager(
             self._conn,
@@ -35,13 +37,6 @@ def _hard_ttl(self, ttl_ms, hard_ttl_ms):
 
     def _key(self, key):
         return f"{self.namespace}:{key}" if self.namespace else key
-        
-    # def lease(self, key, ttl_ms=None, hard_ttl_ms=None):
-    #     key = self._key(key)
-    #     ttl = self._ttl(ttl_ms)
-    #     hard_ttl = self._hard_ttl(ttl,hard_ttl_ms)
-
-    #     return Lease(None, key, ttl, hard_ttl, self._conn)
 
     async def once(self, key, fn, ttl_ms=None, hard_ttl_ms=None, kwargs=None):
         key = self._key(key)

sentinel-py/sentinel/cli.py

@@ -0,0 +1,50 @@
+import os
+import argparse
+import psycopg
+
+try:
+    from dotenv import load_dotenv
+    load_dotenv()
+except ImportError:
+    pass
+
+from .core import inspect
+
+def get_conn():
+    db_url = os.environ.get("DATABASE_URL")
+    if not db_url:
+        raise EnvironmentError(
+            "No DATABASE_URL found. Set it in your environment or .env file.\n"
+            "Example: DATABASE_URL=postgresql://user:pass@localhost/mydb"
+        )
+    return psycopg.connect(db_url)
+
+def main():
+    parser = argparse.ArgumentParser(prog="sen", description="Sentinel CLI")
+    subparsers = parser.add_subparsers(dest="command")
+
+    inspect_parser = subparsers.add_parser("inspect", help="Inspect a lease by key")
+    inspect_parser.add_argument("key", type=str)
+
+    args = parser.parse_args()
+
+    if args.command == "inspect":
+        conn = get_conn()
+        result = inspect(conn, args.key)
+
+        if result is None:
+            print(f"No lease found for key: {args.key}")
+        else:
+            print(f"status:           {result.status}")
+            print(f"lease_alive:      {result.lease_alive}")
+            print(f"owner_id:         {result.owner_id}")
+            print(f"fencing_token:    {result.fencing_token}")
+            print(f"lease_expires_at: {result.lease_expires_at}")
+            print(f"lease_updated_at: {result.lease_updated_at}")
+            print(f"hard_expires_at:  {result.hard_expires_at}")
+            print(f"execution_result: {result.execution_result}")
+    else:
+        parser.print_help()
+
+if __name__ == "__main__":
+    main()

sentinel-py/sentinel/core.py

@@ -1,6 +1,6 @@
 import json
 from .utils import get_owner_id, row_to_dict
-from .result import AcquireResult, OperationResult
+from .result import AcquireResult, OperationResult, InspectResult
 
 def acquire(conn, key, *, owner_id=None, ttl_ms=10000, hard_ttl_ms = None):
 
@@ -182,4 +182,30 @@ def expire_lease(conn, key, *, owner_id, fencing_token):
     conn.commit()
     return OperationResult(success)
 
+def inspect(conn, key):
+    with conn.cursor() as cur:
+        cur.execute("""
+        SELECT owner_id, lease_expires_at, fencing_token, status, 
+        lease_expires_at > NOW() AS lease_alive, lease_updated_at,
+        hard_expires_at, execution_result
+        FROM sentinel_leases
+        WHERE key = %s 
+        """, (key,))
+
+        result = cur.fetchone()
 
+        if result is None:
+            return None
+
+        row = row_to_dict(cur, result)
+
+    return InspectResult(
+        key = key,
+        owner_id = row["owner_id"],
+        fencing_token = row["fencing_token"],
+        status = row["status"],
+        lease_alive = row["lease_alive"],
+        lease_expires_at = row["lease_expires_at"],
+        lease_updated_at = row["lease_updated_at"],
+        hard_expires_at = row["hard_expires_at"],
+        execution_result = row["execution_result"])

sentinel-py/sentinel/once.py

@@ -13,7 +13,6 @@
 
 from .heartbeat_config import get_manager
 from .logging import logger
-from .reconciliation import Reconcile
 from .result import OnceResult
 
 
@@ -36,7 +35,6 @@ def __init__(
         self.kwargs = kwargs or {}
         self.owns_connection = owns_connection
 
-        self.reconcile = Reconcile(get_conn)
         self._task = None
 
     def run(self):
@@ -76,8 +74,7 @@ def run(self):
                         success=False,
                         status="executing",
                         uncertain=True,
-                        execution_alive=False,
-                        reconcile=self.reconcile
+                        execution_alive=False
                     )
 
                 if acquired.status == "executing" and acquired.lease_alive:
@@ -162,8 +159,7 @@ def run(self):
                     status="executing",
                     execution_alive=False,
                     uncertain=True,
-                    exception=e,
-                    reconcile=self.reconcile
+                    exception=e
                 )
 
             # Finalize canonical completion
@@ -184,8 +180,7 @@ def run(self):
                 return OnceResult(
                     success=False,
                     status=completed.status,
-                    uncertain=True,
-                    reconcile=self.reconcile
+                    uncertain=True
                 )
 
             return OnceResult(