tastyware/streaq:v6.0.1

[Graeme22]

What's Changed

• Add Trio support (pending anyio support in coredis) by @Graeme22 in Add Trio support (pending anyio support in coredis) #80
  This is a large change, powered by simultaneous changes to the coredis dependency. Here's a summary of the changes:

  • The library now works with both asyncio and Trio. You can change the backend by passing anyio_backend to workers:

    # asyncio is the default
    worker = Worker(anyio_backend="trio", anyio_kwargs={"restrict_keyboard_interrupt_to_checkpoints": True})

  • streaQ now uses structured concurrency everywhere for more predictability. If you haven't read Nathaniel Smith's article on structured concurrency, go read it now--it's a fantastic explanation of the motivations and an introduction to some of the concepts.
  • As a consequence of SC, async context managers for the worker are back! You'll need to use the context manager for any operation that involves communicating with Redis:

    async with worker:
        task = await sleeper.enqueue()
        print(await task.result(3))

    This means the recommended way of integrating with FastAPI has changed a bit:

    @asynccontextmanager
    async def app_lifespan(app: FastAPI) -> AsyncGenerator[None]:
        async with worker:
            yield
    
    app = FastAPI(lifespan=app_lifespan)

    See here for more details.

• dynamic cron jobs by @Graeme22 in dynamic cron jobs #117
  Aside from defining cron jobs with the decorator, you can also schedule tasks dynamically:

  task = await sleeper.enqueue(1).start(schedule="*/5 * * * *")  # every 5 minutes

  This causes the task to run repeatedly with the given arguments at the given schedule. To stop scheduling a repeating task, you can use:

  await task.unschedule()
  # OR
  await worker.unschedule_by_id(task.id)

• Modular changes by @Graeme22 in Modular changes #125
  Adds new function, Worker.include(), which simply modifies the included worker's tasks to point to it instead. At its face this sounds like it could lead to problems, but since a) Only one worker is running per process at a time; and b) Workers that are included in other workers are not aware of that at import time, practically this is a clean and easy solution. To make this a bit more concrete:

  from streaq import Worker
  
  child = Worker()
  
  @child.task()
  async def foobar() -> None: ...

  from streaq import Worker
  
  from child_module import child
  
  parent = Worker()
  parent.include(child)

  Here, we can actually run either worker independently without issues: streaq run child_module:child and streaq run parent_module:parent should both work fine, because when running just the child, the parent.include(child) line will never run.

  The other element of this PR that is quite crucial is detaching the ability to access task and worker context from the worker instance itself. Under the hood this is done with a couple global ContextVars, but in practice it uses a clean, FastAPI-like dependency injection pattern:

  from streaq import TaskContext, TaskDepends, WorkerDepends
  from typing import NamedTuple
  
  class WorkerContext(NamedTuple):
      db_connection: MyDBDriver
  
  @worker.task()
  async def has_dependencies(
      task_context: TaskContext = TaskDepends(),
      worker_context: WorkerContext = WorkerDepends(),
  ) -> None:
      print(task_context.task_id)
      print(worker_context.db_connection)

  What this enables that was not previously possible is a total separation of concerns. Task definitions can now access dependencies without having to know anything about the top level Worker they will eventually be included in. This replaces the previous Worker.context and Worker.task_context() functions altogether.

• CLI makeover: the streaq command has been split into two subcommands, streaq run and streaq web for running a worker and the web UI respectively. As a bonus this allowed for getting some extra coverage that was previously marked with # pragma: no cover.

• Typing for pipelined tasks is drastically better for both Task.then() and Task.__or__(), with some fancy overloads and a TypeVarTuple that allow for nearly perfect typing of the complex scenario where results from one task are fed into the next.

• Workers no longer attempt to re-enqueue tasks on shutdown--instead, we assume all shutdowns are unsafe, and allow Worker.idle_timeout to handle retrying. The default idle timeout has been decreased from 5 minutes to 1 minute to reflect this change.

• Redis scripts are now Redis functions in a single Lua library file. This means we now use FCALL instead of EVALSHA, which slightly increases performance in pipelines. This also means Redis versions less than 7.0 are no longer supported.

• Task abortion now handles delayed tasks better, aborting them right away instead of marking them and leaving them untouched

• TaskResult now has slightly better typing. Instead of accessing TaskResult.result which was either the task's output or any raised exception, TaskResult.result now only contains the task's output and errors if you access it when TaskResult.success is False. Exceptions are now contained in TaskResult.exception, and errors if you access it when TaskResult.success is True.

  async with worker:
      result = await task.result(3)
      if result.success:
          print(result.result)  # this is present only if success
      else:
          raise result.exception  # this is present only if not success

• Add date in logs by @nsteinmetz in Add date in logs #91

• Ability to set custom formatters for task results and exceptions in Web UI by @espdev in Ability to set custom formatters for task results and exceptions in Web UI #95

• Update docs for SAQ comparison: abort support by @vikigenius in Update docs for SAQ comparison: abort support #100

• Custom worker id feature by @espdev in Custom worker id feature #101

• Place sleep after the first healthcheck, not before it. by @espdev in Place sleep after the first healthcheck, not before it. #104

• Add the info about exception traceback serialization with tblib to the docs by @espdev in Add the info about exception traceback serialization with tblib to the docs #105

• Improve the user experience when running tests by @espdev in Improve the user experience when running tests #107

• StreaqRetry, StreaqError, and StreaqCancelled have been moved from utils to the streaq.types module.

• Worker can now be initialized from a coredis.ConnectionPool instead of a connection URI.

• The Redis health check cron job is now registered at worker startup instead of in the constructor.

• Configure Renovate by @renovate[bot] in Configure Renovate #133

• Redis Sentinel now in test coverage everywhere a worker is used

• Tests now use a Docker compose cluster

• Added "Contributing" page to docs

• Benchmarks updated

• Web UI no longer uses alru_cache since it's not compatible with anyio

New Contributors

• @vikigenius made their first contribution in Update docs for SAQ comparison: abort support #100

Full Changelog: v5.2.2...v6.0.1

---

This discussion was created from the release tastyware/streaq:v6.0.1.