mirror of https://github.com/peter4431/rq.git
Multiple results using Redis Streams (#1725)
* WIP job results * Result can now be saved * Successfully saved and restored result * result.save() should accept pipeline * Successful results are saved * Failures are now saved properly too. * Added test for Result.get_latest() * Checkpoint * Got Result.all() to work * Added Result.count(), Result.delete() * Backward compatibility for job.result and job.exc_info * Added some typing * More typing stuff * Fixed typing in job.py * More typing updates * Only keep the last 10 results * Documented job.results() * Got results test to pass * Don't run test_results.py on Redis server < 5.0 * Fixed mock import on some Python versions * Remove Redis 3 from test matrix * Jobs should never use the new Result implementation if server is < 5.0 * Results should only be created is Redis stream is supported. * Added back Redis 3 to test matrix * Fixed job.supports_redis_streams * Fixed worker test * Updated docs.main
Selwin Ong
2 years ago
committed by
GitHub
parent
09856f9924
commit
0691b4d46e
@ -0,0 +1,173 @@
|
|||||||
|
import json
|
||||||
|
from typing import Any, Optional
|
||||||
|
import zlib
|
||||||
|
|
||||||
|
from base64 import b64decode, b64encode
|
||||||
|
from datetime import datetime, timezone
|
||||||
|
from enum import Enum
|
||||||
|
from uuid import uuid4
|
||||||
|
|
||||||
|
from redis import Redis
|
||||||
|
from redis.client import Pipeline
|
||||||
|
|
||||||
|
from .compat import decode_redis_hash
|
||||||
|
from .job import Job
|
||||||
|
from .serializers import resolve_serializer
|
||||||
|
from .utils import now
|
||||||
|
|
||||||
|
|
||||||
|
def get_key(job_id):
|
||||||
|
return 'rq:results:%s' % job_id
|
||||||
|
|
||||||
|
|
||||||
|
class Result(object):
|
||||||
|
|
||||||
|
class Type(Enum):
|
||||||
|
SUCCESSFUL = 1
|
||||||
|
FAILED = 2
|
||||||
|
STOPPED = 3
|
||||||
|
|
||||||
|
def __init__(self, job_id: str, type: Type, connection: Redis, id: Optional[str] = None,
|
||||||
|
created_at: Optional[datetime] = None, return_value: Optional[Any] = None,
|
||||||
|
exc_string: Optional[str] = None, serializer=None):
|
||||||
|
self.return_value = return_value
|
||||||
|
self.exc_string = exc_string
|
||||||
|
self.type = type
|
||||||
|
self.created_at = created_at if created_at else now()
|
||||||
|
self.serializer = resolve_serializer(serializer)
|
||||||
|
self.connection = connection
|
||||||
|
self.job_id = job_id
|
||||||
|
self.id = id
|
||||||
|
|
||||||
|
def __repr__(self):
|
||||||
|
return f'Result(id={self.id}, type={self.Type(self.type).name})'
|
||||||
|
|
||||||
|
def __eq__(self, other):
|
||||||
|
try:
|
||||||
|
return self.id == other.id
|
||||||
|
except AttributeError:
|
||||||
|
return False
|
||||||
|
|
||||||
|
def __bool__(self):
|
||||||
|
return bool(self.id)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def create(cls, job, type, ttl, return_value=None, exc_string=None, pipeline=None):
|
||||||
|
result = cls(job_id=job.id, type=type, connection=job.connection,
|
||||||
|
return_value=return_value,
|
||||||
|
exc_string=exc_string, serializer=job.serializer)
|
||||||
|
result.save(ttl=ttl, pipeline=pipeline)
|
||||||
|
return result
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def create_failure(cls, job, ttl, exc_string, pipeline=None):
|
||||||
|
result = cls(job_id=job.id, type=cls.Type.FAILED, connection=job.connection,
|
||||||
|
exc_string=exc_string, serializer=job.serializer)
|
||||||
|
result.save(ttl=ttl, pipeline=pipeline)
|
||||||
|
return result
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def all(cls, job: Job, serializer=None):
|
||||||
|
"""Returns all results for job"""
|
||||||
|
# response = job.connection.zrange(cls.get_key(job.id), 0, 10, desc=True, withscores=True)
|
||||||
|
response = job.connection.xrevrange(cls.get_key(job.id), '+', '-')
|
||||||
|
results = []
|
||||||
|
for (result_id, payload) in response:
|
||||||
|
results.append(
|
||||||
|
cls.restore(job.id, result_id.decode(), payload,
|
||||||
|
connection=job.connection, serializer=serializer)
|
||||||
|
)
|
||||||
|
|
||||||
|
return results
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def count(cls, job: Job) -> int:
|
||||||
|
"""Returns the number of job results"""
|
||||||
|
return job.connection.xlen(cls.get_key(job.id))
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def delete_all(cls, job: Job) -> None:
|
||||||
|
"""Delete all job results"""
|
||||||
|
job.connection.delete(cls.get_key(job.id))
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def restore(cls, job_id: str, result_id: str, payload: dict, connection: Redis, serializer=None) -> 'Result':
|
||||||
|
"""Create a Result object from given Redis payload"""
|
||||||
|
created_at = datetime.fromtimestamp(
|
||||||
|
int(result_id.split('-')[0]) / 1000, tz=timezone.utc
|
||||||
|
)
|
||||||
|
payload = decode_redis_hash(payload)
|
||||||
|
# data, timestamp = payload
|
||||||
|
# result_data = json.loads(data)
|
||||||
|
# created_at = datetime.fromtimestamp(timestamp, tz=timezone.utc)
|
||||||
|
|
||||||
|
serializer = resolve_serializer(serializer)
|
||||||
|
return_value = payload.get('return_value')
|
||||||
|
if return_value is not None:
|
||||||
|
return_value = serializer.loads(b64decode(return_value.decode()))
|
||||||
|
|
||||||
|
exc_string = payload.get('exc_string')
|
||||||
|
if exc_string:
|
||||||
|
exc_string = zlib.decompress(b64decode(exc_string)).decode()
|
||||||
|
|
||||||
|
return Result(job_id, Result.Type(int(payload['type'])), connection=connection,
|
||||||
|
id=result_id,
|
||||||
|
created_at=created_at,
|
||||||
|
return_value=return_value,
|
||||||
|
exc_string=exc_string)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def fetch(cls, job: Job, serializer=None) -> Optional['Result']:
|
||||||
|
"""Fetch a result that matches a given job ID. The current sorted set
|
||||||
|
based implementation does not allow us to fetch a given key by ID
|
||||||
|
so we need to iterate through results, deserialize the payload and
|
||||||
|
look for a matching ID.
|
||||||
|
|
||||||
|
Future Redis streams based implementation may make this more efficient
|
||||||
|
and scalable.
|
||||||
|
"""
|
||||||
|
return None
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def fetch_latest(cls, job: Job, serializer=None) -> Optional['Result']:
|
||||||
|
"""Returns the latest result for given job instance or ID"""
|
||||||
|
# response = job.connection.zrevrangebyscore(cls.get_key(job.id), '+inf', '-inf',
|
||||||
|
# start=0, num=1, withscores=True)
|
||||||
|
response = job.connection.xrevrange(cls.get_key(job.id), '+', '-', count=1)
|
||||||
|
if not response:
|
||||||
|
return None
|
||||||
|
|
||||||
|
result_id, payload = response[0]
|
||||||
|
return cls.restore(job.id, result_id.decode(), payload,
|
||||||
|
connection=job.connection, serializer=serializer)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def get_key(cls, job_id):
|
||||||
|
return 'rq:results:%s' % job_id
|
||||||
|
|
||||||
|
def save(self, ttl, pipeline=None):
|
||||||
|
"""Save result data to Redis"""
|
||||||
|
key = self.get_key(self.job_id)
|
||||||
|
|
||||||
|
connection = pipeline if pipeline is not None else self.connection
|
||||||
|
# result = connection.zadd(key, {self.serialize(): self.created_at.timestamp()})
|
||||||
|
result = connection.xadd(key, self.serialize(), maxlen=10)
|
||||||
|
# If xadd() is called in a pipeline, it returns a pipeline object instead of stream ID
|
||||||
|
if pipeline is None:
|
||||||
|
self.id = result.decode()
|
||||||
|
if ttl is not None:
|
||||||
|
connection.expire(key, ttl)
|
||||||
|
return self.id
|
||||||
|
|
||||||
|
def serialize(self):
|
||||||
|
data = {'type': self.type.value}
|
||||||
|
|
||||||
|
if self.exc_string is not None:
|
||||||
|
data['exc_string'] = b64encode(zlib.compress(self.exc_string.encode())).decode()
|
||||||
|
|
||||||
|
serialized = self.serializer.dumps(self.return_value)
|
||||||
|
if self.return_value is not None:
|
||||||
|
data['return_value'] = b64encode(serialized).decode()
|
||||||
|
|
||||||
|
# return json.dumps(data)
|
||||||
|
return data
|
||||||
@ -0,0 +1,201 @@
|
|||||||
|
import unittest
|
||||||
|
|
||||||
|
from datetime import timedelta
|
||||||
|
from unittest.mock import patch, PropertyMock
|
||||||
|
|
||||||
|
from redis import Redis
|
||||||
|
|
||||||
|
from tests import RQTestCase
|
||||||
|
|
||||||
|
from rq.job import Job
|
||||||
|
from rq.queue import Queue
|
||||||
|
from rq.registry import StartedJobRegistry
|
||||||
|
from rq.results import Result, get_key
|
||||||
|
from rq.utils import get_version, utcnow
|
||||||
|
from rq.worker import Worker
|
||||||
|
|
||||||
|
from .fixtures import say_hello
|
||||||
|
|
||||||
|
|
||||||
|
@unittest.skipIf(get_version(Redis()) < (5, 0, 0), 'Skip if Redis server < 5.0')
|
||||||
|
class TestScheduledJobRegistry(RQTestCase):
|
||||||
|
|
||||||
|
def test_save_and_get_result(self):
|
||||||
|
"""Ensure data is saved properly"""
|
||||||
|
queue = Queue(connection=self.connection)
|
||||||
|
job = queue.enqueue(say_hello)
|
||||||
|
|
||||||
|
result = Result.fetch_latest(job)
|
||||||
|
self.assertIsNone(result)
|
||||||
|
|
||||||
|
Result.create(job, Result.Type.SUCCESSFUL, ttl=10, return_value=1)
|
||||||
|
result = Result.fetch_latest(job)
|
||||||
|
self.assertEqual(result.return_value, 1)
|
||||||
|
self.assertEqual(job.latest_result().return_value, 1)
|
||||||
|
|
||||||
|
# Check that ttl is properly set
|
||||||
|
key = get_key(job.id)
|
||||||
|
ttl = self.connection.pttl(key)
|
||||||
|
self.assertTrue(5000 < ttl <= 10000)
|
||||||
|
|
||||||
|
# Check job with None return value
|
||||||
|
Result.create(job, Result.Type.SUCCESSFUL, ttl=10, return_value=None)
|
||||||
|
result = Result.fetch_latest(job)
|
||||||
|
self.assertIsNone(result.return_value)
|
||||||
|
Result.create(job, Result.Type.SUCCESSFUL, ttl=10, return_value=2)
|
||||||
|
result = Result.fetch_latest(job)
|
||||||
|
self.assertEqual(result.return_value, 2)
|
||||||
|
|
||||||
|
def test_create_failure(self):
|
||||||
|
"""Ensure data is saved properly"""
|
||||||
|
queue = Queue(connection=self.connection)
|
||||||
|
job = queue.enqueue(say_hello)
|
||||||
|
Result.create_failure(job, ttl=10, exc_string='exception')
|
||||||
|
result = Result.fetch_latest(job)
|
||||||
|
self.assertEqual(result.exc_string, 'exception')
|
||||||
|
|
||||||
|
# Check that ttl is properly set
|
||||||
|
key = get_key(job.id)
|
||||||
|
ttl = self.connection.pttl(key)
|
||||||
|
self.assertTrue(5000 < ttl <= 10000)
|
||||||
|
|
||||||
|
def test_getting_results(self):
|
||||||
|
"""Check getting all execution results"""
|
||||||
|
queue = Queue(connection=self.connection)
|
||||||
|
job = queue.enqueue(say_hello)
|
||||||
|
|
||||||
|
# latest_result() returns None when there's no result
|
||||||
|
self.assertIsNone(job.latest_result())
|
||||||
|
|
||||||
|
result_1 = Result.create_failure(job, ttl=10, exc_string='exception')
|
||||||
|
result_2 = Result.create(job, Result.Type.SUCCESSFUL, ttl=10, return_value=1)
|
||||||
|
result_3 = Result.create(job, Result.Type.SUCCESSFUL, ttl=10, return_value=1)
|
||||||
|
|
||||||
|
# Result.fetch_latest() returns the latest result
|
||||||
|
result = Result.fetch_latest(job)
|
||||||
|
self.assertEqual(result, result_3)
|
||||||
|
self.assertEqual(job.latest_result(), result_3)
|
||||||
|
|
||||||
|
# Result.all() and job.results() returns all results, newest first
|
||||||
|
results = Result.all(job)
|
||||||
|
self.assertEqual(results, [result_3, result_2, result_1])
|
||||||
|
self.assertEqual(job.results(), [result_3, result_2, result_1])
|
||||||
|
|
||||||
|
def test_count(self):
|
||||||
|
"""Result.count(job) returns number of results"""
|
||||||
|
queue = Queue(connection=self.connection)
|
||||||
|
job = queue.enqueue(say_hello)
|
||||||
|
self.assertEqual(Result.count(job), 0)
|
||||||
|
Result.create_failure(job, ttl=10, exc_string='exception')
|
||||||
|
self.assertEqual(Result.count(job), 1)
|
||||||
|
Result.create(job, Result.Type.SUCCESSFUL, ttl=10, return_value=1)
|
||||||
|
self.assertEqual(Result.count(job), 2)
|
||||||
|
|
||||||
|
def test_delete_all(self):
|
||||||
|
"""Result.delete_all(job) deletes all results from Redis"""
|
||||||
|
queue = Queue(connection=self.connection)
|
||||||
|
job = queue.enqueue(say_hello)
|
||||||
|
Result.create_failure(job, ttl=10, exc_string='exception')
|
||||||
|
Result.create(job, Result.Type.SUCCESSFUL, ttl=10, return_value=1)
|
||||||
|
Result.delete_all(job)
|
||||||
|
self.assertEqual(Result.count(job), 0)
|
||||||
|
|
||||||
|
def test_job_successful_result_fallback(self):
|
||||||
|
"""Changes to job.result handling should be backwards compatible."""
|
||||||
|
queue = Queue(connection=self.connection)
|
||||||
|
job = queue.enqueue(say_hello)
|
||||||
|
worker = Worker([queue])
|
||||||
|
worker.register_birth()
|
||||||
|
|
||||||
|
self.assertEqual(worker.failed_job_count, 0)
|
||||||
|
self.assertEqual(worker.successful_job_count, 0)
|
||||||
|
self.assertEqual(worker.total_working_time, 0)
|
||||||
|
|
||||||
|
# These should only run on workers that supports Redis streams
|
||||||
|
registry = StartedJobRegistry(connection=self.connection)
|
||||||
|
job.started_at = utcnow()
|
||||||
|
job.ended_at = job.started_at + timedelta(seconds=0.75)
|
||||||
|
job._result = 'Success'
|
||||||
|
worker.handle_job_success(job, queue, registry)
|
||||||
|
|
||||||
|
payload = self.connection.hgetall(job.key)
|
||||||
|
self.assertFalse(b'result' in payload.keys())
|
||||||
|
self.assertEqual(job.result, 'Success')
|
||||||
|
|
||||||
|
with patch('rq.worker.Worker.supports_redis_streams', new_callable=PropertyMock) as mock:
|
||||||
|
mock.return_value = False
|
||||||
|
worker = Worker([queue])
|
||||||
|
worker.register_birth()
|
||||||
|
job = queue.enqueue(say_hello)
|
||||||
|
job._result = 'Success'
|
||||||
|
job.started_at = utcnow()
|
||||||
|
job.ended_at = job.started_at + timedelta(seconds=0.75)
|
||||||
|
|
||||||
|
# If `save_result_to_job` = True, result will be saved to job
|
||||||
|
# hash, simulating older versions of RQ
|
||||||
|
|
||||||
|
worker.handle_job_success(job, queue, registry)
|
||||||
|
payload = self.connection.hgetall(job.key)
|
||||||
|
self.assertTrue(b'result' in payload.keys())
|
||||||
|
# Delete all new result objects so we only have result stored in job hash,
|
||||||
|
# this should simulate a job that was executed in an earlier RQ version
|
||||||
|
self.assertEqual(job.result, 'Success')
|
||||||
|
|
||||||
|
def test_job_failed_result_fallback(self):
|
||||||
|
"""Changes to job.result failure handling should be backwards compatible."""
|
||||||
|
queue = Queue(connection=self.connection)
|
||||||
|
job = queue.enqueue(say_hello)
|
||||||
|
worker = Worker([queue])
|
||||||
|
worker.register_birth()
|
||||||
|
|
||||||
|
self.assertEqual(worker.failed_job_count, 0)
|
||||||
|
self.assertEqual(worker.successful_job_count, 0)
|
||||||
|
self.assertEqual(worker.total_working_time, 0)
|
||||||
|
|
||||||
|
registry = StartedJobRegistry(connection=self.connection)
|
||||||
|
job.started_at = utcnow()
|
||||||
|
job.ended_at = job.started_at + timedelta(seconds=0.75)
|
||||||
|
worker.handle_job_failure(job, exc_string='Error', queue=queue,
|
||||||
|
started_job_registry=registry)
|
||||||
|
|
||||||
|
job = Job.fetch(job.id, connection=self.connection)
|
||||||
|
payload = self.connection.hgetall(job.key)
|
||||||
|
self.assertFalse(b'exc_info' in payload.keys())
|
||||||
|
self.assertEqual(job.exc_info, 'Error')
|
||||||
|
|
||||||
|
with patch('rq.worker.Worker.supports_redis_streams', new_callable=PropertyMock) as mock:
|
||||||
|
mock.return_value = False
|
||||||
|
worker = Worker([queue])
|
||||||
|
worker.register_birth()
|
||||||
|
|
||||||
|
job = queue.enqueue(say_hello)
|
||||||
|
job.started_at = utcnow()
|
||||||
|
job.ended_at = job.started_at + timedelta(seconds=0.75)
|
||||||
|
|
||||||
|
# If `save_result_to_job` = True, result will be saved to job
|
||||||
|
# hash, simulating older versions of RQ
|
||||||
|
|
||||||
|
worker.handle_job_failure(job, exc_string='Error', queue=queue,
|
||||||
|
started_job_registry=registry)
|
||||||
|
payload = self.connection.hgetall(job.key)
|
||||||
|
self.assertTrue(b'exc_info' in payload.keys())
|
||||||
|
# Delete all new result objects so we only have result stored in job hash,
|
||||||
|
# this should simulate a job that was executed in an earlier RQ version
|
||||||
|
Result.delete_all(job)
|
||||||
|
job = Job.fetch(job.id, connection=self.connection)
|
||||||
|
self.assertEqual(job.exc_info, 'Error')
|
||||||
|
|
||||||
|
def test_job_return_value(self):
|
||||||
|
"""Test job.return_value"""
|
||||||
|
queue = Queue(connection=self.connection)
|
||||||
|
job = queue.enqueue(say_hello)
|
||||||
|
|
||||||
|
# Returns None when there's no result
|
||||||
|
self.assertIsNone(job.return_value())
|
||||||
|
|
||||||
|
Result.create(job, Result.Type.SUCCESSFUL, ttl=10, return_value=1)
|
||||||
|
self.assertEqual(job.return_value(), 1)
|
||||||
|
|
||||||
|
# Returns None if latest result is a failure
|
||||||
|
Result.create_failure(job, ttl=10, exc_string='exception')
|
||||||
|
self.assertIsNone(job.return_value(refresh=True))
|
||||||
Loading…
Reference in New Issue