SQLAlchemy store should remove finished schedules on cleanup

src/apscheduler/datastores/sqlalchemy.py

@@ -963,7 +963,7 @@ async def cleanup(self) -> None:
                     results = await self._execute(conn, query)
                     if finished_schedule_ids := dict(results.all()):
                         delete = self._t_schedules.delete().where(
-                            ~self._t_schedules.c.id.in_(finished_schedule_ids)
+                            self._t_schedules.c.id.in_(finished_schedule_ids)
                         )
                         await self._execute(conn, delete)
 

tests/test_schedulers.py

@@ -689,13 +689,16 @@ def check_contextvars() -> None:
 
     async def test_explicit_cleanup(self, raw_datastore: DataStore) -> None:
         send, receive = create_memory_object_stream[Event](1)
-        async with AsyncScheduler(cleanup_interval=None) as scheduler:
+        async with AsyncScheduler(raw_datastore, cleanup_interval=None) as scheduler:
             scheduler.subscribe(send.send, {ScheduleRemoved})
+            event = anyio.Event()
+
+            scheduler.subscribe(lambda _: event.set(), {JobReleased}, one_shot=True)
+
             await scheduler.start_in_background()
 
             # Add a job whose result expires after 1 ms
-            event = anyio.Event()
-            job_id = await scheduler.add_job(event.set, result_expiration_time=0.001)
+            job_id = await scheduler.add_job(dummy_async_job, result_expiration_time=0.001)
             with fail_after(3):
                 await event.wait()
 
@@ -708,8 +711,11 @@ async def test_explicit_cleanup(self, raw_datastore: DataStore) -> None:
 
             # Add a schedule to immediately set the event
             event = anyio.Event()
+
+            scheduler.subscribe(lambda _: event.set(), {JobReleased}, one_shot=True)
+
             await scheduler.add_schedule(
-                event.set, DateTrigger(datetime.now(timezone.utc)), id="event_set"
+                dummy_async_job, DateTrigger(datetime.now(timezone.utc)), id="event_set"
             )
             with fail_after(3):
                 await event.wait()