Make backend public and fix all type errors

- Change _backend to backend (make it public, not protected)
- Worker legitimately needs backend access - tight coupling is OK
- Fix Backend.lock() protocol signature (remove decorator, return AsyncContextManager)
- Add type: ignore for uuid_extensions import (external lib without stubs)
- Update all references from docket._backend to docket.backend
- Update tests to use public backend

All type errors in docket.py, worker.py, backends/base.py, and
execution.py are now resolved (0 errors in modified files).

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: zzstoatzz

src/docket/backends/base.py

@@ -6,10 +6,9 @@
 
 from __future__ import annotations
 
-from contextlib import asynccontextmanager
 from datetime import datetime
 from typing import (
-    AsyncGenerator,
+    AsyncContextManager,
     Collection,
     Protocol,
     Sequence,
@@ -278,22 +277,21 @@ async def release_concurrency_slot(
         ...
 
     # Distributed lock operations
-    @asynccontextmanager
-    async def lock(
+    def lock(
         self,
         key: str,
         timeout: float,
         blocking: bool = True,
-    ) -> AsyncGenerator[None, None]:
+    ) -> AsyncContextManager[None]:
         """Acquire a distributed lock.
 
         Args:
             key: Lock key
             timeout: Lock timeout in seconds
             blocking: Whether to block waiting for lock
 
-        Yields:
-            None when lock is acquired
+        Returns:
+            Async context manager that yields when lock is acquired
 
         Raises:
             LockError: If lock cannot be acquired (when blocking=False)

src/docket/docket.py

@@ -21,7 +21,7 @@
 )
 
 from opentelemetry import trace
-from uuid_extensions import uuid7
+from uuid_extensions import uuid7  # type: ignore[import-untyped]
 
 from .backends import Backend, RedisBackend
 from .execution import (
@@ -95,9 +95,9 @@ async def my_task(greeting: str, recipient: str) -> None:
 
     tasks: dict[str, TaskFunction]
     strike_list: StrikeList
+    backend: Backend
 
     _monitor_strikes_task: asyncio.Task[None]
-    _backend: Backend
 
     def __init__(
         self,
@@ -124,7 +124,7 @@ def __init__(
         self.heartbeat_interval = heartbeat_interval
         self.missed_heartbeats = missed_heartbeats
         # Create backend based on URL scheme
-        self._backend = RedisBackend(name, url)
+        self.backend = RedisBackend(name, url)
 
     @property
     def worker_group_name(self) -> str:
@@ -136,7 +136,7 @@ async def __aenter__(self) -> Self:
         self.tasks = {fn.__name__: fn for fn in standard_tasks}
         self.strike_list = StrikeList()
 
-        await self._backend.initialize()
+        await self.backend.initialize()
         self._monitor_strikes_task = asyncio.create_task(self._monitor_strikes())
 
         return self
@@ -159,7 +159,7 @@ async def __aexit__(
             except asyncio.CancelledError:
                 pass
 
-        await self._backend.close()
+        await self.backend.close()
 
     def register(self, function: TaskFunction) -> None:
         """Register a task with the Docket.
@@ -265,7 +265,7 @@ async def scheduler(*args: P.args, **kwargs: P.kwargs) -> Execution:
                 )
                 return execution
 
-            await self._backend.schedule_task(execution, replace=False)
+            await self.backend.schedule_task(execution, replace=False)
 
             TASKS_ADDED.add(1, {**self.labels(), **execution.general_labels()})
             TASKS_SCHEDULED.add(1, {**self.labels(), **execution.general_labels()})
@@ -340,7 +340,7 @@ async def scheduler(*args: P.args, **kwargs: P.kwargs) -> Execution:
                 )
                 return execution
 
-            await self._backend.schedule_task(execution, replace=True)
+            await self.backend.schedule_task(execution, replace=True)
 
             TASKS_REPLACED.add(1, {**self.labels(), **execution.general_labels()})
             TASKS_CANCELLED.add(1, {**self.labels(), **execution.general_labels()})
@@ -376,7 +376,7 @@ async def schedule(self, execution: Execution) -> None:
                 "code.function.name": execution.function.__name__,
             },
         ):
-            await self._backend.schedule_task(execution, replace=False)
+            await self.backend.schedule_task(execution, replace=False)
 
         TASKS_SCHEDULED.add(1, {**self.labels(), **execution.general_labels()})
 
@@ -390,30 +390,30 @@ async def cancel(self, key: str) -> None:
             "docket.cancel",
             attributes={**self.labels(), "docket.key": key},
         ):
-            await self._backend.cancel_task(key)
+            await self.backend.cancel_task(key)
 
         TASKS_CANCELLED.add(1, self.labels())
 
     @property
     def queue_key(self) -> str:
-        return self._backend.queue_key
+        return self.backend.queue_key
 
     @property
     def stream_key(self) -> str:
-        return self._backend.stream_key
+        return self.backend.stream_key
 
     def known_task_key(self, key: str) -> str:
-        return self._backend.known_task_key(key)
+        return self.backend.known_task_key(key)
 
     def parked_task_key(self, key: str) -> str:
-        return self._backend.parked_task_key(key)
+        return self.backend.parked_task_key(key)
 
     def stream_id_key(self, key: str) -> str:
-        return self._backend.stream_id_key(key)
+        return self.backend.stream_id_key(key)
 
     @property
     def strike_key(self) -> str:
-        return self._backend.strike_key
+        return self.backend.strike_key
 
     async def strike(
         self,
@@ -469,14 +469,14 @@ async def _send_strike_instruction(self, instruction: StrikeInstruction) -> None
                 **instruction.labels(),
             },
         ):
-            await self._backend.send_strike_instruction(instruction)
+            await self.backend.send_strike_instruction(instruction)
             self.strike_list.update(instruction)
 
     async def _monitor_strikes(self) -> NoReturn:
         last_id = "0-0"
         while True:
             try:
-                instructions = await self._backend.receive_strike_instructions(
+                instructions = await self.backend.receive_strike_instructions(
                     last_id, timeout_ms=60_000
                 )
                 for message_id, instruction in instructions:
@@ -518,7 +518,7 @@ async def snapshot(self) -> DocketSnapshot:
             total_tasks,
             future_executions,
             running_executions,
-        ) = await self._backend.snapshot(self.tasks)
+        ) = await self.backend.snapshot(self.tasks)
         workers = await self.workers()
 
         return DocketSnapshot(
@@ -527,13 +527,13 @@ async def snapshot(self) -> DocketSnapshot:
 
     @property
     def workers_set(self) -> str:
-        return self._backend.workers_set
+        return self.backend.workers_set
 
     def worker_tasks_set(self, worker_name: str) -> str:
-        return self._backend.worker_tasks_set(worker_name)
+        return self.backend.worker_tasks_set(worker_name)
 
     def task_workers_set(self, task_name: str) -> str:
-        return self._backend.task_workers_set(task_name)
+        return self.backend.task_workers_set(task_name)
 
     async def workers(self) -> Collection[WorkerInfo]:
         """Get a list of all workers that have sent heartbeats to the Docket.
@@ -544,7 +544,7 @@ async def workers(self) -> Collection[WorkerInfo]:
         heartbeat_timeout = (
             self.heartbeat_interval.total_seconds() * self.missed_heartbeats
         )
-        return await self._backend.get_workers(heartbeat_timeout)
+        return await self.backend.get_workers(heartbeat_timeout)
 
     async def task_workers(self, task_name: str) -> Collection[WorkerInfo]:
         """Get a list of all workers that are able to execute a given task.
@@ -558,7 +558,7 @@ async def task_workers(self, task_name: str) -> Collection[WorkerInfo]:
         heartbeat_timeout = (
             self.heartbeat_interval.total_seconds() * self.missed_heartbeats
         )
-        return await self._backend.get_task_workers(task_name, heartbeat_timeout)
+        return await self.backend.get_task_workers(task_name, heartbeat_timeout)
 
     async def clear(self) -> int:
         """Clear all pending and scheduled tasks from the docket.
@@ -574,4 +574,4 @@ async def clear(self) -> int:
             "docket.clear",
             attributes=self.labels(),
         ):
-            return await self._backend.clear()
+            return await self.backend.clear()

src/docket/worker.py

@@ -245,11 +245,11 @@ async def _worker_loop(self, forever: bool = False):
 
         async def check_for_work() -> bool:
             logger.debug("Checking for work", extra=log_context)
-            return await self.docket._backend.check_for_work()
+            return await self.docket.backend.check_for_work()
 
         async def get_redeliveries() -> RedisReadGroupResponse:
             logger.debug("Getting redeliveries", extra=log_context)
-            redeliveries = await self.docket._backend.reclaim_stale_tasks(
+            redeliveries = await self.docket.backend.reclaim_stale_tasks(
                 worker_name=self.name,
                 max_count=available_slots,
                 stale_timeout_ms=int(self.redelivery_timeout.total_seconds() * 1000),
@@ -258,7 +258,7 @@ async def get_redeliveries() -> RedisReadGroupResponse:
 
         async def get_new_deliveries() -> RedisReadGroupResponse:
             logger.debug("Getting new deliveries", extra=log_context)
-            messages = await self.docket._backend.claim_tasks(
+            messages = await self.docket.backend.claim_tasks(
                 worker_name=self.name,
                 max_count=available_slots,
                 timeout_ms=int(self.minimum_check_interval.total_seconds() * 1000),
@@ -301,7 +301,7 @@ async def process_completed_tasks() -> None:
 
         async def ack_message(message_id: RedisMessageID) -> None:
             logger.debug("Acknowledging message", extra=log_context)
-            await self.docket._backend.acknowledge_task(message_id)
+            await self.docket.backend.acknowledge_task(message_id)
 
         has_work: bool = True
 
@@ -364,7 +364,7 @@ async def _scheduler_loop(
             try:
                 logger.debug("Scheduling due tasks", extra=log_context)
                 now = datetime.now(timezone.utc)
-                total_work, due_work = await self.docket._backend.move_due_tasks(now)
+                total_work, due_work = await self.docket.backend.move_due_tasks(now)
 
                 if due_work > 0:
                     logger.debug(
@@ -388,7 +388,7 @@ async def _scheduler_loop(
 
     async def _schedule_all_automatic_perpetual_tasks(self) -> None:
         try:
-            async with self.docket._backend.lock(
+            async with self.docket.backend.lock(
                 f"{self.docket.name}:perpetual:lock", timeout=10, blocking=False
             ):
                 for task_function in self.docket.tasks.values():
@@ -419,7 +419,7 @@ async def _delete_known_task(
             return
 
         logger.debug("Deleting known task", extra=self._log_context())
-        await self.docket._backend.delete_known_task(key)
+        await self.docket.backend.delete_known_task(key)
 
     async def _execute(self, execution: Execution) -> None:
         log_context = {**self._log_context(), **execution.specific_labels()}
@@ -486,7 +486,7 @@ async def _execute(self, execution: Execution) -> None:
                             # Try to acquire a concurrency slot
                             expiry_seconds = self.redelivery_timeout.total_seconds()
                             acquired = (
-                                await self.docket._backend.acquire_concurrency_slot(
+                                await self.docket.backend.acquire_concurrency_slot(
                                     key=concurrency_key,
                                     max_concurrent=concurrency_limit.max_concurrent,
                                     worker_name=self.name,
@@ -615,7 +615,7 @@ async def _execute(self, execution: Execution) -> None:
                             scope = concurrency_limit.scope or self.docket.name
                             concurrency_key = f"{scope}:concurrency:{concurrency_limit.argument_name}:{argument_value}"
 
-                            await self.docket._backend.release_concurrency_slot(
+                            await self.docket.backend.release_concurrency_slot(
                                 key=concurrency_key,
                                 worker_name=self.name,
                                 task_key=execution.key,
@@ -740,7 +740,7 @@ async def _heartbeat(self) -> None:
                     stream_depth,
                     overdue_depth,
                     schedule_depth,
-                ) = await self.docket._backend.record_heartbeat(
+                ) = await self.docket.backend.record_heartbeat(
                     worker_name=self.name,
                     task_names=task_names,
                     timestamp=now,

tests/test_docket.py

@@ -150,14 +150,14 @@ async def test_clear_no_redis_key_leaks(docket: Docket, the_task: AsyncMock):
     await docket.add(the_task, when=future)("scheduled1")
     await docket.add(the_task, when=future + timedelta(seconds=1))("scheduled2")
 
-    async with docket._backend.redis() as r:
+    async with docket.backend.redis() as r:
         keys_before = cast(list[str], await r.keys("*"))  # type: ignore
         keys_before_count = len(keys_before)
 
     result = await docket.clear()
     assert result == 5
 
-    async with docket._backend.redis() as r:
+    async with docket.backend.redis() as r:
         keys_after = cast(list[str], await r.keys("*"))  # type: ignore
         keys_after_count = len(keys_after)