A task always has a Blueprint

but to configure an task, their blueprint needs to be an App

procrastinate/blueprints.py

@@ -86,10 +86,12 @@ def _check_stack(self):
             )
 
     def _register_task(self, task: "tasks.Task") -> None:
+        from procrastinate import tasks
+
         # Each call to _add_task may raise TaskAlreadyRegistered.
         # We're using an intermediary dict to make sure that if the registration
         # is interrupted midway though, self.tasks is left unmodified.
-        to_add = {}
+        to_add: Dict[str, tasks.Task] = {}
         self._add_task(task=task, name=task.name, to=to_add)
 
         for alias in task.aliases:
@@ -150,9 +152,11 @@ def add_tasks_from(self, blueprint: "Blueprint", *, namespace: str) -> None:
             raise exceptions.TaskAlreadyRegistered(
                 f"A namespace named {namespace} was already registered"
             )
-        # Modify existing tasks and other blueprint to add the namespace
+        # Modify existing tasks and other blueprint to add the namespace, and
+        # set the blueprint
         for task in set(blueprint.tasks.values()):
             task.add_namespace(namespace)
+            task.blueprint = self
         blueprint.tasks = new_tasks
 
         # Finally, add the namespaced tasks to this namespace
@@ -226,7 +230,7 @@ def _wrap(func: Callable[..., "tasks.Task"]):
 
             task = tasks.Task(
                 func,
-                app=None,
+                blueprint=self,
                 queue=queue,
                 lock=lock,
                 queueing_lock=queueing_lock,

procrastinate/tasks.py

@@ -2,7 +2,8 @@
 import logging
 from typing import Any, Callable, Dict, List, Optional
 
-from procrastinate import app, exceptions, jobs, manager
+from procrastinate import app as app_module
+from procrastinate import blueprints, exceptions, jobs, manager
 from procrastinate import retry as retry_module
 from procrastinate import types, utils
 
@@ -72,7 +73,7 @@ def __init__(
         self,
         func: Callable,
         *,
-        app: Optional[app.App],
+        blueprint: blueprints.Blueprint,
         # task naming
         name: Optional[str] = None,
         aliases: Optional[List[str]] = None,
@@ -85,7 +86,7 @@ def __init__(
         queueing_lock: Optional[str] = None,
     ):
         self.queue = queue
-        self.app = app
+        self.blueprint = blueprint
         self.func: Callable = func
         self.aliases = aliases if aliases else []
         self.retry_strategy = retry_module.get_retry_strategy(retry)
@@ -177,12 +178,14 @@ def configure(
         ValueError
             If you try to define both schedule_at and schedule_in
         """
-        if self.app is None:
+        if not isinstance(self.blueprint, app_module.App):
             raise exceptions.UnboundTaskError
 
+        app = self.blueprint
+
         return configure_task(
             name=self.name,
-            job_manager=self.app.job_manager,
+            job_manager=app.job_manager,
             lock=lock if lock is not None else self.lock,
             queueing_lock=(
                 queueing_lock if queueing_lock is not None else self.queueing_lock