later

Common errors

• RuntimeWarning: coroutine 'my_coroutine' was never awaited

  cause Attempting to run an asynchronous coroutine function without properly awaiting it, often seen in `unittest` methods that are not declared `async`.
  fix If this error occurs in a test method, ensure the method is defined as `async def test_my_feature(self):`. If elsewhere, ensure you are using `await` or `asyncio.create_task()` to schedule the coroutine.

• TypeError: object NoneType can't be used in 'await' expression

  cause This typically happens when an asynchronous function or operation that is expected to return a coroutine (and thus be awaited) actually returns `None` due to an error, early exit, or incorrect mock setup.
  fix Inspect the function or mock returning `None`. Ensure all paths return a coroutine, a Future, or a Task. When mocking `async` context managers with `later.unittest.mock.AsyncContextManager`, confirm the mock correctly provides `__aenter__` and `__aexit__` methods.

• Task exception was never retrieved

  cause An `asyncio` task raised an unhandled exception, and no code explicitly `await`ed the task to retrieve its result or handle the exception. This is a common `asyncio` footgun.
  fix When creating tasks (e.g., with `asyncio.create_task()` or `later.as_task`), always ensure they are `await`ed at some point (e.g., via `asyncio.gather()`, `later.Watcher.join()`, or by explicitly `await task`) to propagate exceptions or retrieve results.

Warnings

• gotcha Directly cancelling asyncio.Task objects using `task.cancel()` can lead to unawaited tasks and resource leaks if not handled carefully, especially in complex scenarios or when tasks create sub-tasks.
  Fix: Always prefer `from later import cancel` when cancelling tasks. `later.cancel` ensures that the task's cancellation is awaited, preventing orphaned tasks and allowing for proper cleanup.
• gotcha When using `later.unittest.TestCase`, ensure your test methods are `async` functions. Forgetting `async def` will cause the test runner to treat them as regular synchronous methods, leading to `RuntimeWarning` for unawaited coroutines or incorrect test behavior.
  Fix: Declare all test methods within `later.unittest.TestCase` as `async def test_something(self):` to ensure proper `asyncio` test execution and task management.
• breaking While specific breaking changes are not extensively documented publicly for minor versions, being an `asyncio` utility, internal changes to `asyncio` in new Python versions (e.g., Python 3.11, 3.12, 3.13) could potentially necessitate updates in `later`'s implementation or usage patterns, particularly for advanced features interacting directly with the `asyncio` event loop.
  Fix: Always test `later` components thoroughly when upgrading to a new major Python version. Consult the `later` GitHub repository's issues and release notes for compatibility information with new Python releases.

Install

• pip install later Install latest version

Imports

• TestCase

  from later.unittest import TestCase

  A specialized asyncio.IsolatedAsyncioTestCase that ensures tasks are not orphaned.
• AsyncContextManager

  from later.unittest.mock import AsyncContextManager

  A factory for easily mocking AsyncContextManager instances in tests.
• cancel

  from later import cancel

  The recommended way to cancel an asyncio.Task/Future and ensure it is awaited properly.
• as_task

  from later import as_task

  Decorator to turn coroutines into asyncio.Tasks automatically.
• Watcher

  from later import Watcher

  A task management utility similar to asyncio.TaskGroup, designed to watch tasks and act upon their completion or failure.
• herd

  from later import herd

  A decorator for coroutines providing basic thundering herd protection.
• BiDirectionalEvent

  from later.event import BiDirectionalEvent

  A two-way asyncio.Event for handshake-style synchronization between coroutines.

Quickstart

This quickstart demonstrates using `later.Watcher` to manage `asyncio` tasks and `later.cancel` for safe task cancellation. It spawns two worker coroutines, cancels one after a delay, and then waits for all managed tasks to finish.

import asyncio
from later import as_task, Watcher, cancel

async def worker(name, delay):
    print(f"{name}: Starting work...")
    try:
        await asyncio.sleep(delay)
        print(f"{name}: Work done.")
    except asyncio.CancelledError:
        print(f"{name}: Work cancelled.")

async def main():
    watcher = Watcher()
    task1 = await watcher.spawn(worker('Worker 1', 2))
    task2 = await watcher.spawn(worker('Worker 2', 5))

    await asyncio.sleep(1) # Let tasks start
    print("Main: Cancelling Worker 1")
    await cancel(task1)

    # Wait for all tasks in the watcher to complete or be cancelled
    await watcher.join()
    print("Main: All tasks managed by Watcher have completed.")

if __name__ == "__main__":
    asyncio.run(main())