Implement Butler._query.

python/lsst/daf/butler/direct_butler.py

@@ -65,6 +65,7 @@
 from ._timespan import Timespan
 from .datastore import Datastore, NullDatastore
 from .dimensions import DataCoordinate, Dimension
+from .direct_query_driver import DirectQueryDriver
 from .progress import Progress
 from .queries import Query
 from .registry import (
@@ -2102,7 +2103,12 @@ def dimensions(self) -> DimensionUniverse:
     @contextlib.contextmanager
     def _query(self) -> Iterator[Query]:
         # Docstring inherited.
-        raise NotImplementedError("TODO DM-41159")
+        driver = DirectQueryDriver(
+            self._registry._db, self.dimensions, self._registry._managers, self._registry.defaults
+        )
+        query = Query(driver)
+        with driver:
+            yield query
 
     def _preload_cache(self) -> None:
         """Immediately load caches that are used for common operations."""