support eager_start and **kwargs in create_task, matching CPython 3.11–3.14

loop.pyx.in is a new Tempita template that emits the correct
create_task signature for each Python version at build time:
- 3.14+: `def create_task(self, coro, **kwargs)` forwarding all
  kwargs to the factory/Task, matching CPython gh-128552
- 3.13:  `def create_task(self, coro, *, name=None, context=None, **kwargs)`
  passing extra kwargs through to the factory and Task
- 3.11/3.12: `def create_task(self, coro, *, name=None, context=None)`
  matching CPython's behaviour for those versions
- <3.11:  legacy context.run path

setup.py uses Cython.Tempita to render loop.pyx.in → loop.pyx before
calling cythonize, using the process_tempita pattern.

Fixes: #746

Author: graingert

setup.py

@@ -94,8 +94,8 @@ def finalize_options(self):
 
         for extension in self.distribution.ext_modules:
             for i, sfile in enumerate(extension.sources):
-                if sfile.endswith('.pyx'):
-                    prefix, ext = os.path.splitext(sfile)
+                if sfile.endswith('.pyx') or sfile.endswith('.pyx.in'):
+                    prefix = sfile[:-len('.pyx.in')] if sfile.endswith('.pyx.in') else sfile[:-len('.pyx')]
                     cfile = prefix + '.c'
 
                     if os.path.exists(cfile) and not self.cython_always:
@@ -128,8 +128,24 @@ def finalize_options(self):
                         CYTHON_DEPENDENCY, Cython.__version__
                     ))
 
+            from Cython import Tempita
             from Cython.Build import cythonize
 
+            def process_tempita(source):
+                assert source.endswith('.pyx.in')
+                with open(source) as f:
+                    rendered = Tempita.sub(f.read())
+                out = source[:-len('.in')]  # .pyx.in -> .pyx
+                with open(out, 'w') as f:
+                    f.write(rendered)
+                return out
+
+            for extension in self.distribution.ext_modules:
+                extension.sources = [
+                    process_tempita(s) if s.endswith('.pyx.in') else s
+                    for s in extension.sources
+                ]
+
             directives = {}
             if self.cython_directives:
                 for directive in self.cython_directives.split(','):
@@ -247,7 +263,7 @@ def build_extensions(self):
         Extension(
             "uvloop.loop",
             sources=[
-                "uvloop/loop.pyx",
+                "uvloop/loop.pyx.in",
             ],
             extra_compile_args=MODULES_CFLAGS
         ),

tests/test_base.py

@@ -581,8 +581,8 @@ def factory(loop, coro, **kwargs):
             task = MyTask(coro, loop=loop, **kwargs)
             # Python moved the responsibility to set the name to the Task
             # class constructor, so MyTask.set_name is never called by
-            # Python's create_task.  Compensate for that here.
-            if self.is_asyncio_loop() and "name" in kwargs:
+            # create_task when name is passed as a kwarg.  Compensate here.
+            if "name" in kwargs:
                 task.set_name(kwargs["name"])
             return task
 
@@ -604,6 +604,87 @@ def factory(loop, coro, **kwargs):
         self.loop.set_task_factory(None)
         self.assertIsNone(self.loop.get_task_factory())
 
+    @unittest.skipUnless(sys.version_info >= (3, 14), "requires Python 3.14+")
+    def test_create_task_eager_start_false(self):
+        # create_task must forward **kwargs to the task factory/constructor
+        # so callers can pass eager_start=False (or True) per-task.
+        self.loop._process_events = mock.Mock()
+
+        eager_start_values = []
+
+        class TrackingTask(asyncio.Task):
+            def __init__(self, coro, *, loop, eager_start=False, **kwargs):
+                eager_start_values.append(eager_start)
+                super().__init__(coro, loop=loop, eager_start=eager_start, **kwargs)
+
+        async def coro():
+            pass
+
+        # Without a factory: kwargs go straight to Task.__init__
+        task = self.loop.create_task(coro(), eager_start=False)
+        self.loop.run_until_complete(task)
+
+        # With a factory that accepts **kwargs
+        self.loop.set_task_factory(
+            lambda loop, coro, **kwargs: TrackingTask(coro, loop=loop, **kwargs)
+        )
+        task = self.loop.create_task(coro(), eager_start=False)
+        self.loop.run_until_complete(task)
+        self.assertEqual(eager_start_values[-1], False)
+
+        task = self.loop.create_task(coro(), eager_start=True)
+        self.loop.run_until_complete(task)
+        self.assertEqual(eager_start_values[-1], True)
+
+        self.loop.set_task_factory(None)
+
+    @unittest.skipUnless(sys.version_info >= (3, 12), "requires Python 3.12+")
+    @unittest.skipIf(sys.version_info >= (3, 14), "3.14+ tested separately")
+    def test_eager_task_factory_313(self):
+        # On 3.12/3.13, eager_task_factory can be used with uvloop and
+        # tasks are started eagerly (eager_start=True) by default.
+        self.loop._process_events = mock.Mock()
+
+        eager_start_values = []
+
+        class TrackingTask(asyncio.Task):
+            def __init__(self, coro, *, loop, eager_start=False, **kwargs):
+                eager_start_values.append(eager_start)
+                super().__init__(coro, loop=loop, eager_start=eager_start, **kwargs)
+
+        async def coro():
+            pass
+
+        self.loop.set_task_factory(
+            asyncio.create_eager_task_factory(TrackingTask)
+        )
+        task = self.loop.create_task(coro())
+        self.loop.run_until_complete(task)
+        self.assertEqual(eager_start_values[-1], True)
+
+        # create_eager_task_factory-based factory also accepts eager_start=False
+        # via the factory signature — but that must be triggered through the
+        # factory directly, not via create_task (which doesn't forward it on
+        # pre-3.14).  Verify the factory itself works with eager_start=False.
+        factory = asyncio.create_eager_task_factory(TrackingTask)
+        task = factory(self.loop, coro(), eager_start=False)
+        self.loop.run_until_complete(task)
+        self.assertEqual(eager_start_values[-1], False)
+
+        self.loop.set_task_factory(None)
+
+    @unittest.skipIf(sys.version_info >= (3, 14), "3.14+ supports **kwargs")
+    def test_create_task_eager_start_raises_pre314(self):
+        # On pre-3.14, create_task does not forward eager_start, matching
+        # CPython's BaseEventLoop.create_task signature on those versions.
+        self.loop._process_events = mock.Mock()
+
+        async def coro():
+            pass
+
+        with self.assertRaises(TypeError):
+            self.loop.create_task(coro(), eager_start=False)
+
     def test_shutdown_asyncgens_01(self):
         finalized = list()
 

uvloop/loop.pyx

@@ -1404,46 +1404,20 @@ cdef class Loop:
         """Create a Future object attached to the loop."""
         return self._new_future()
 
-    def create_task(self, coro, *, name=None, context=None):
+    def create_task(self, coro, **kwargs):
         """Schedule a coroutine object.
 
         Return a task object.
-
-        If name is not None, task.set_name(name) will be called if the task
-        object has the set_name attribute, true for default Task in CPython.
-
-        An optional keyword-only context argument allows specifying a custom
-        contextvars.Context for the coro to run in. The current context copy is
-        created when no context is provided.
         """
         self._check_closed()
-        if PY311:
-            if self._task_factory is None:
-                task = aio_Task(coro, loop=self, context=context)
-            else:
-                task = self._task_factory(self, coro, context=context)
-        else:
-            if context is None:
-                if self._task_factory is None:
-                    task = aio_Task(coro, loop=self)
-                else:
-                    task = self._task_factory(self, coro)
-            else:
-                if self._task_factory is None:
-                    task = context.run(aio_Task, coro, self)
-                else:
-                    task = context.run(self._task_factory, self, coro)
-
-        # copied from asyncio.tasks._set_task_name (bpo-34270)
-        if name is not None:
-            try:
-                set_name = task.set_name
-            except AttributeError:
-                pass
-            else:
-                set_name(name)
+        if self._task_factory is not None:
+            return self._task_factory(self, coro, **kwargs)
+        task = aio_Task(coro, loop=self, **kwargs)
+        try:
+            return task
+        finally:
+            del task
 
-        return task
 
     def set_task_factory(self, factory):
         """Set a task factory that will be used by loop.create_task().