Merge pull request #360 from phenobarbital/dev

bump version of background service

Author: phenobarbital

navigator/background/tracker/models.py

@@ -19,6 +19,7 @@ class JobRecord(BaseModel):
     """
     task_id: str = Field(default=gen_uuid)
     name: str = None
+    content: Optional[str] = None
     status: str = 'pending'
     attributes: Dict[str, Any] = Field(default_factory=dict)
     result: Optional[Any] = None

navigator/background/tracker/redis.py

@@ -8,6 +8,7 @@
 
 
 Encoder = Callable[[Any], bytes]
+MAX_TTL = 30 * 24 * 3600
 
 
 class RedisJobTracker:
@@ -17,7 +18,12 @@ class RedisJobTracker:
     key  <prefix>{task_id}
     and keeps a Set <prefix>__all   with the ids for quick listing.
     """
-    def __init__(self, url: str = None, prefix: str = 'job:') -> None:
+    def __init__(
+        self,
+        url: str = None,
+        prefix: str = 'job:',
+        ttl_seconds: int = 30 * 24 * 3600,  # 30 days
+    ) -> None:
         self._url = url or CACHE_URL
         self._redis: redis.Redis = redis.from_url(
             self._url,
@@ -28,6 +34,7 @@ def __init__(self, url: str = None, prefix: str = 'job:') -> None:
         self._decoder: Encoder = self._decode_model
         self.prefix = prefix if prefix.endswith(":") else f"{prefix}:"
         self._lock = asyncio.Lock()
+        self._ttl = min(ttl_seconds, MAX_TTL)
 
     def _key(self, task_id: str) -> str:
         return f"{self.prefix}{task_id}"
@@ -49,7 +56,9 @@ async def create_job(self, job: JobRecord, **kwargs) -> JobRecord:
             ) from exc
         key = self._key(job.task_id)
         async with self._lock:
-            await self._redis.set(key, self._encoder(job))
+            await self._redis.set(
+                key, self._encoder(job), ex=self._ttl
+            )
             await self._redis.sadd(self._set_key, job.task_id)
 
             # Create secondary index for attributes if provided
@@ -75,7 +84,9 @@ async def _update(self, job_id: str, **patch) -> None:
             rec: JobRecord = self._decoder(payload)
             for k, v in patch.items():
                 setattr(rec, k, v)
-            await self._redis.set(key, self._encoder(rec))
+            await self._redis.set(
+                key, self._encoder(rec), keepttl=True
+            )  # keep the TTL
 
             # Update secondary index for attributes if they are part of the patch
             if 'attributes' in patch:
@@ -178,7 +189,11 @@ async def forget(self, job_id: str) -> None:
             if payload:
                 rec: JobRecord = self._decoder(payload)
                 for k, v in rec.attributes.items():
-                    await self._redis.srem(self._attr_key(k, v), job_id)
+                    akey = self._attr_key(k, v)
+                    await self._redis.srem(akey, job_id)
+                    # set attr-set to expire in 24 h if now empty
+                    if await self._redis.scard(akey) == 0:
+                        await self._redis.expire(akey, 86400)
 
             await self._redis.delete(self._key(job_id))
             await self._redis.srem(self._set_key, job_id)

navigator/background/wrappers/__init__.py

@@ -63,8 +63,6 @@ def __init__(
         retry_delay: float = 0.0,
         **kwargs
     ):
-        self.args = args
-        self.kwargs = kwargs
         self.fn = fn
         self.tracker = tracker
         self._name: str = kwargs.pop('name', fn.__name__ if fn else 'unknown_task')
@@ -77,16 +75,27 @@ def __init__(
             )
         self.jitter: float = jitter
         # Create the Job Record at status "pending"
+        # generate a list of arguments accepted by JobRecord:
+        job_args = {}
+        content = kwargs.pop('content', None)
+        for k, v in kwargs.items():
+            if not k.startswith('_'):
+                if k in JobRecord.__fields__:
+                    job_args[k] = v
         self.job_record: JobRecord = JobRecord(
             name=self._name,
+            content=content,
             status=job_status,
-            **kwargs
+            **job_args
         )
         self.logger = logger or logging.getLogger('NAV.Queue.TaskWrapper')
         # Retry information:
         self.max_retries = max_retries
         self.retries_done = 0
         self.retry_delay = retry_delay
+        # Store the arguments and keyword arguments for the Function
+        self.args = args
+        self.kwargs = kwargs
 
     @property
     def task_uuid(self) -> uuid.UUID:

navigator/version.py

@@ -4,7 +4,7 @@
 __description__ = (
     "Navigator Web Framework based on aiohttp, " "with batteries included."
 )
-__version__ = "2.13.0"
+__version__ = "2.13.1"
 __copyright__ = "Copyright (c) 2020-2024 Jesus Lara"
 __author__ = "Jesus Lara"
 __author_email__ = "jesuslarag@gmail.com"