Repository: samuelcolvin/arq

Issue:
How to get jobs in order to abort
Hi and thanks for this work!

In order to abort some deferred jobs, I need to get an instance of `arq.jobs.Job` but I don't find how to get jobs.
I'm using FastAPI with an endpoint to enqueue job and another to abort job, so I don't have the job instance returns by `enqueue_job()`.

How can I do that?

Thanks

Discussion:
codecov[bot]: # [Codecov](https://codecov.io/gh/samuelcolvin/arq/pull/378?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin) Report
> Merging [#378](https://codecov.io/gh/samuelcolvin/arq/pull/378?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin) (405e4fd) into [main](https://codecov.io/gh/samuelcolvin/arq/commit/bfa8e68fff7063e792f4a5deef0f778eb549d73c?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin) (bfa8e68) will **decrease** coverage by `0.28%`.
> The diff coverage is `100.00%`.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main     #378      +/-   ##
==========================================
- Coverage   98.66%   98.38%   -0.29%     
==========================================
  Files          11       11              
  Lines        1052     1050       -2     
  Branches      199      170      -29     
==========================================
- Hits         1038     1033       -5     
- Misses          6        7       +1     
- Partials        8       10       +2     
```


| [Impacted Files](https://codecov.io/gh/samuelcolvin/arq/pull/378?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin) | Coverage Δ | |
|---|---|---|
| [arq/worker.py](https://codecov.io/gh/samuelcolvin/arq/pull/378/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin#diff-YXJxL3dvcmtlci5weQ==) | `98.96% <ø> (-0.21%)` | :arrow_down: |
| [arq/connections.py](https://codecov.io/gh/samuelcolvin/arq/pull/378/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin#diff-YXJxL2Nvbm5lY3Rpb25zLnB5) | `95.20% <100.00%> (ø)` | |
| [arq/jobs.py](https://codecov.io/gh/samuelcolvin/arq/pull/378/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin#diff-YXJxL2pvYnMucHk=) | `98.15% <100.00%> (ø)` | |
| [arq/typing.py](https://codecov.io/gh/samuelcolvin/arq/pull/378/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin#diff-YXJxL3R5cGluZy5weQ==) | `90.47% <0.00%> (-9.53%)` | :arrow_down: |
| [arq/cron.py](https://codecov.io/gh/samuelcolvin/arq/pull/378/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin#diff-YXJxL2Nyb24ucHk=) | `100.00% <0.00%> (ø)` | |
| [arq/utils.py](https://codecov.io/gh/samuelcolvin/arq/pull/378/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin#diff-YXJxL3V0aWxzLnB5) | `100.00% <0.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/samuelcolvin/arq/pull/378?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/samuelcolvin/arq/pull/378?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin). Last update [bfa8e68...405e4fd](https://codecov.io/gh/samuelcolvin/arq/pull/378?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Samuel+Colvin).

</details>
JonasKs: Hi @gerazenobi 😊
It needs a review from @samuelcolvin, who's a busy man these days (pydantic v2 is around the corner). I have no estimate, but I'd assume some time after v2 is out 😊
gerazenobi: @JonasKs thanks for the quick answer ! 

Just to confirm: there is no workaround to get from queued job to actual job correct ?  (without doing any custom tracking on my end)
gerazenobi: @JonasKs indeed, however [I need to be able to enqueue multiple jobs of the "same nature"](https://github.com/samuelcolvin/arq/issues/394) hence I can't use custom_ids unfortunately: once an id is used then we can't delete the job nor re-enqueue the job again.

Patch:
diff --git a/arq/connections.py b/arq/connections.py
index d4fc443..69ac8ce 100644
--- a/arq/connections.py
+++ b/arq/connections.py
@@ -193,6 +193,7 @@ class ArqRedis(BaseRedis):
         assert v is not None, f'job "{key}" not found'
         jd = deserialize_job(v, deserializer=self.job_deserializer)
         jd.score = score
+        jd.job_id = job_id.decode()
         return jd
 
     async def queued_jobs(self, *, queue_name: Optional[str] = None) -> List[JobDef]:
diff --git a/arq/jobs.py b/arq/jobs.py
index 8028cbe..d0c0a5e 100644
--- a/arq/jobs.py
+++ b/arq/jobs.py
@@ -47,6 +47,7 @@ class JobDef:
     job_try: int
     enqueue_time: datetime
     score: Optional[int]
+    job_id: Optional[str]
 
     def __post_init__(self) -> None:
         if isinstance(self.score, float):
@@ -60,7 +61,6 @@ class JobResult(JobDef):
     start_time: datetime
     finish_time: datetime
     queue_name: str
-    job_id: Optional[str] = None
 
 
 class Job:
@@ -238,6 +238,7 @@ def serialize_result(
     finished_ms: int,
     ref: str,
     queue_name: str,
+    job_id: str,
     *,
     serializer: Optional[Serializer] = None,
 ) -> Optional[bytes]:
@@ -252,6 +253,7 @@ def serialize_result(
         'st': start_ms,
         'ft': finished_ms,
         'q': queue_name,
+        'id': job_id,
     }
     if serializer is None:
         serializer = pickle.dumps
@@ -281,6 +283,7 @@ def deserialize_job(r: bytes, *, deserializer: Optional[Deserializer] = None) ->
             job_try=d['t'],
             enqueue_time=ms_to_datetime(d['et']),
             score=None,
+            job_id=None,
         )
     except Exception as e:
         raise DeserializationError('unable to deserialize job') from e
@@ -315,6 +318,7 @@ def deserialize_result(r: bytes, *, deserializer: Optional[Deserializer] = None)
             start_time=ms_to_datetime(d['st']),
             finish_time=ms_to_datetime(d['ft']),
             queue_name=d.get('q', '<unknown>'),
+            job_id=d.get('id', '<unknown>'),
         )
     except Exception as e:
         raise DeserializationError('unable to deserialize job result') from e
diff --git a/arq/worker.py b/arq/worker.py
index 81afd5b..7ff5393 100644
--- a/arq/worker.py
+++ b/arq/worker.py
@@ -236,7 +236,11 @@ class Worker:
         self.on_job_start = on_job_start
         self.on_job_end = on_job_end
         self.after_job_end = after_job_end
-        self.sem = asyncio.BoundedSemaphore(max_jobs)
+
+        self.max_jobs = max_jobs
+        self.sem = asyncio.BoundedSemaphore(max_jobs + 1)
+        self.job_counter: int = 0
+
         self.job_timeout_s = to_seconds(job_timeout)
         self.keep_result_s = to_seconds(keep_result)
         self.keep_result_forever = keep_result_forever
@@ -374,13 +378,13 @@ class Worker:
                 return
             count = min(burst_jobs_remaining, count)
         if self.allow_pick_jobs:
-            async with self.sem:  # don't bother with zrangebyscore until we have "space" to run the jobs
+            if self.job_counter < self.max_jobs:
                 now = timestamp_ms()
                 job_ids = await self.pool.zrangebyscore(
                     self.queue_name, min=float('-inf'), start=self._queue_read_offset, num=count, max=now
                 )
 
-            await self.start_jobs(job_ids)
+                await self.start_jobs(job_ids)
 
         if self.allow_abort_jobs:
             await self._cancel_aborted_jobs()
@@ -419,12 +423,23 @@ class Worker:
             self.aborting_tasks.update(aborted)
             await self.pool.zrem(abort_jobs_ss, *aborted)
 
+    def _release_sem_dec_counter_on_complete(self) -> None:
+        self.job_counter = self.job_counter - 1
+        self.sem.release()
+
     async def start_jobs(self, job_ids: List[bytes]) -> None:
         """
         For each job id, get the job definition, check it's not running and start it in a task
         """
         for job_id_b in job_ids:
             await self.sem.acquire()
+
+            if self.job_counter >= self.max_jobs:
+                self.sem.release()
+                return None
+
+            self.job_counter = self.job_counter + 1
+
             job_id = job_id_b.decode()
             in_progress_key = in_progress_key_prefix + job_id
             async with self.pool.pipeline(transaction=True) as pipe:
@@ -433,6 +448,7 @@ class Worker:
                 score = await pipe.zscore(self.queue_name, job_id)
                 if ongoing_exists or not score:
                     # job already started elsewhere, or already finished and removed from queue
+                    self.job_counter = self.job_counter - 1
                     self.sem.release()
                     logger.debug('job %s already running elsewhere', job_id)
                     continue
@@ -445,11 +461,12 @@ class Worker:
                     await pipe.execute()
                 except (ResponseError, WatchError):
                     # job already started elsewhere since we got 'existing'
+                    self.job_counter = self.job_counter - 1
                     self.sem.release()
                     logger.debug('multi-exec error, job %s already started elsewhere', job_id)
                 else:
                     t = self.loop.create_task(self.run_job(job_id, int(score)))
-                    t.add_done_callback(lambda _: self.sem.release())
+                    t.add_done_callback(lambda _: self._release_sem_dec_counter_on_complete())
                     self.tasks[job_id] = t
 
     async def run_job(self, job_id: str, score: int) -> None:  # noqa: C901
@@ -484,6 +501,7 @@ class Worker:
                 ref=f'{job_id}:{function_name}',
                 serializer=self.job_serializer,
                 queue_name=self.queue_name,
+                job_id=job_id,
             )
             await asyncio.shield(self.finish_failed_job(job_id, result_data_))
 
@@ -539,6 +557,7 @@ class Worker:
                 timestamp_ms(),
                 ref,
                 self.queue_name,
+                job_id=job_id,
                 serializer=self.job_serializer,
             )
             return await asyncio.shield(self.finish_failed_job(job_id, result_data))
@@ -632,6 +651,7 @@ class Worker:
                 finished_ms,
                 ref,
                 self.queue_name,
+                job_id=job_id,
                 serializer=self.job_serializer,
             )
 
diff --git a/docs/examples/job_ids.py b/docs/examples/job_ids.py
index 9de9b8c..1148591 100644
--- a/docs/examples/job_ids.py
+++ b/docs/examples/job_ids.py
@@ -2,6 +2,8 @@ import asyncio
 
 from arq import create_pool
 from arq.connections import RedisSettings
+from arq.jobs import Job
+
 
 async def the_task(ctx):
     print('running the task with id', ctx['job_id'])
@@ -37,6 +39,14 @@ async def main():
     >  None
     """
 
+    # you can retrieve jobs by using arq.jobs.Job
+    await redis.enqueue_job('the_task', _job_id='my_job')
+    job5 = Job(job_id='my_job', redis=redis)
+    print(job5)
+    """
+    <arq job my_job>
+    """
+
 class WorkerSettings:
     functions = [the_task]
 
diff --git a/setup.py b/setup.py
deleted file mode 100644
index 94cd7b8..0000000
--- a/setup.py
+++ /dev/null
@@ -1,28 +0,0 @@
-import sys
-
-sys.stderr.write(
-    """
-===============================
-Unsupported installation method
-===============================
-arq no longer supports installation with `python setup.py install`.
-Please use `python -m pip install .` instead.
-"""
-)
-sys.exit(1)
-
-
-# The below code will never execute, however GitHub is particularly
-# picky about where it finds Python packaging metadata.
-# See: https://github.com/github/feedback/discussions/6456
-#
-# To be removed once GitHub catches up.
-
-setup(
-    name='arq',
-    install_requires=[
-        'redis[hiredis]>=4.2.0',
-        'click>=8.0',
-        'typing-extensions>=4.1.0',
-    ],
-)


Test patch:
diff --git a/tests/test_jobs.py b/tests/test_jobs.py
index 634a8b0..f8f6c8c 100644
--- a/tests/test_jobs.py
+++ b/tests/test_jobs.py
@@ -2,7 +2,7 @@ import asyncio
 import pickle
 
 import pytest
-from dirty_equals import IsNow
+from dirty_equals import IsNow, IsStr
 
 from arq import Worker, func
 from arq.connections import ArqRedis, RedisSettings, create_pool
@@ -89,6 +89,7 @@ async def test_enqueue_job(arq_redis: ArqRedis, worker, queue_name=default_queue
         finish_time=IsNow(tz='utc'),
         score=None,
         queue_name=expected_queue_name,
+        job_id=IsStr(),
     )
     results = await arq_redis.all_job_results()
     assert results == [
@@ -139,9 +140,9 @@ async def test_cant_unpickle_at_all():
         def __getstate__(self):
             raise TypeError("this doesn't pickle")
 
-    r1 = serialize_result('foobar', (1,), {}, 1, 123, True, Foobar(), 123, 123, 'testing', 'test-queue')
+    r1 = serialize_result('foobar', (1,), {}, 1, 123, True, Foobar(), 123, 123, 'testing', 'test-queue', 'job_1')
     assert isinstance(r1, bytes)
-    r2 = serialize_result('foobar', (Foobar(),), {}, 1, 123, True, Foobar(), 123, 123, 'testing', 'test-queue')
+    r2 = serialize_result('foobar', (Foobar(),), {}, 1, 123, True, Foobar(), 123, 123, 'testing', 'test-queue', 'job_1')
     assert r2 is None
 
 
@@ -154,7 +155,19 @@ async def test_custom_serializer():
         return b'0123456789'
 
     r1 = serialize_result(
-        'foobar', (1,), {}, 1, 123, True, Foobar(), 123, 123, 'testing', 'test-queue', serializer=custom_serializer
+        'foobar',
+        (1,),
+        {},
+        1,
+        123,
+        True,
+        Foobar(),
+        123,
+        123,
+        'testing',
+        'test-queue',
+        'job_1',
+        serializer=custom_serializer,
     )
     assert r1 == b'0123456789'
     r2 = serialize_result(
@@ -169,6 +182,7 @@ async def test_custom_serializer():
         123,
         'testing',
         'test-queue',
+        'job_1',
         serializer=custom_serializer,
     )
     assert r2 == b'0123456789'
@@ -213,7 +227,7 @@ async def test_get_job_result(arq_redis: ArqRedis):
 
 async def test_result_pole_delay_dep(arq_redis: ArqRedis):
     j = Job('foobar', arq_redis)
-    r = serialize_result('foobar', (1,), {}, 1, 123, True, 42, 123, 123, 'testing', 'test-queue')
+    r = serialize_result('foobar', (1,), {}, 1, 123, True, 42, 123, 123, 'testing', 'test-queue', 'job_1')
     await arq_redis.set(result_key_prefix + j.job_id, r)
     with pytest.warns(
         DeprecationWarning, match='"pole_delay" is deprecated, use the correct spelling "poll_delay" instead'
diff --git a/tests/test_main.py b/tests/test_main.py
index 7c3a983..198c815 100644
--- a/tests/test_main.py
+++ b/tests/test_main.py
@@ -238,11 +238,11 @@ async def test_cant_pickle_result(arq_redis: ArqRedis, worker):
 
 
 async def test_get_jobs(arq_redis: ArqRedis):
-    await arq_redis.enqueue_job('foobar', a=1, b=2, c=3)
+    await arq_redis.enqueue_job('foobar', a=1, b=2, c=3, _job_id='1')
     await asyncio.sleep(0.01)
-    await arq_redis.enqueue_job('second', 4, b=5, c=6)
+    await arq_redis.enqueue_job('second', 4, b=5, c=6, _job_id='2')
     await asyncio.sleep(0.01)
-    await arq_redis.enqueue_job('third', 7, b=8)
+    await arq_redis.enqueue_job('third', 7, b=8, _job_id='3')
     jobs = await arq_redis.queued_jobs()
     assert [dataclasses.asdict(j) for j in jobs] == [
         {
@@ -252,6 +252,7 @@ async def test_get_jobs(arq_redis: ArqRedis):
             'job_try': None,
             'enqueue_time': IsNow(tz='utc'),
             'score': IsInt(),
+            'job_id': '1',
         },
         {
             'function': 'second',
@@ -260,6 +261,7 @@ async def test_get_jobs(arq_redis: ArqRedis):
             'job_try': None,
             'enqueue_time': IsNow(tz='utc'),
             'score': IsInt(),
+            'job_id': '2',
         },
         {
             'function': 'third',
@@ -268,6 +270,7 @@ async def test_get_jobs(arq_redis: ArqRedis):
             'job_try': None,
             'enqueue_time': IsNow(tz='utc'),
             'score': IsInt(),
+            'job_id': '3',
         },
     ]
     assert jobs[0].score < jobs[1].score < jobs[2].score
diff --git a/tests/test_worker.py b/tests/test_worker.py
index aa56085..23dd91d 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -984,6 +984,36 @@ async def test_on_job(arq_redis: ArqRedis, worker):
     assert result['called'] == 4
 
 
+async def test_job_cancel_on_max_jobs(arq_redis: ArqRedis, worker, caplog):
+    async def longfunc(ctx):
+        await asyncio.sleep(3600)
+
+    async def wait_and_abort(job, delay=0.1):
+        await asyncio.sleep(delay)
+        assert await job.abort() is True
+
+    caplog.set_level(logging.INFO)
+    await arq_redis.zadd(abort_jobs_ss, {b'foobar': int(1e9)})
+    job = await arq_redis.enqueue_job('longfunc', _job_id='testing')
+
+    worker: Worker = worker(
+        functions=[func(longfunc, name='longfunc')], allow_abort_jobs=True, poll_delay=0.1, max_jobs=1
+    )
+    assert worker.jobs_complete == 0
+    assert worker.jobs_failed == 0
+    assert worker.jobs_retried == 0
+    await asyncio.gather(wait_and_abort(job), worker.main())
+    await worker.main()
+    assert worker.jobs_complete == 0
+    assert worker.jobs_failed == 1
+    assert worker.jobs_retried == 0
+    log = re.sub(r'\d+.\d\ds', 'X.XXs', '\n'.join(r.message for r in caplog.records))
+    assert 'X.XXs → testing:longfunc()\n  X.XXs ⊘ testing:longfunc aborted' in log
+    assert worker.aborting_tasks == set()
+    assert worker.tasks == {}
+    assert worker.job_tasks == {}
+
+
 async def test_worker_timezone_defaults_to_system_timezone(worker):
     worker = worker(functions=[func(foobar)])
     assert worker.timezone is not None