Adds upstream templating

src/ploomber/spec/dagspec.py

@@ -872,8 +872,8 @@ def process_tasks(dag, dag_spec, root_path=None):
         task, up = task_dict.to_task(dag)
 
         if isinstance(task, TaskGroup):
-            for t in task:
-                upstream_raw[t] = up
+            for t, u in zip(task, up):
+                upstream_raw[t] = u
         else:
             if extract_prod:
                 logger.debug(

src/ploomber/spec/taskspec.py

@@ -351,22 +351,20 @@ def to_task(self, dag):
             else:
                 name_arg = dict(name=name)
 
-            return (
-                TaskGroup.from_grid(
-                    task_class=task_class,
-                    product_class=product_class,
-                    product_primitive=product,
-                    task_kwargs=data,
-                    dag=dag,
-                    grid=grid,
-                    resolve_relative_to=self.project_root,
-                    on_render=on_render,
-                    on_finish=on_finish,
-                    on_failure=on_failure,
-                    params=params,
-                    **name_arg,
-                ),
-                upstream,
+            return TaskGroup.from_grid(
+                task_class=task_class,
+                product_class=product_class,
+                product_primitive=product,
+                task_kwargs=data,
+                dag=dag,
+                grid=grid,
+                resolve_relative_to=self.project_root,
+                on_render=on_render,
+                on_finish=on_finish,
+                on_failure=on_failure,
+                params=params,
+                upstream=upstream,
+                **name_arg,
             )
         else:
             return (

src/ploomber/tasks/taskgroup.py

@@ -95,6 +95,7 @@ def from_params(
         params_array,
         name=None,
         namer=None,
+        upstream=None,
         resolve_relative_to=None,
         on_render=None,
         on_finish=None,
@@ -169,6 +170,7 @@ def from_params(
         # should have a way to extract which placeholders exist?
 
         tasks_all = []
+        upstreams = []
 
         for index, params in enumerate(params_array):
             # each task should get a different deep copy, primarily cause they
@@ -221,6 +223,16 @@ def from_params(
                     f"{product_class} is not supported"
                 )
 
+            # templatize upstream
+            _upstream = []
+            if upstream:
+                for up in upstream:
+                    _upstream.append(Template(
+                        up,
+                        variable_start_string="[[",
+                        variable_end_string="]]",
+                    ).render(**params))
+
             t = task_class(
                 product=product, dag=dag, name=task_name, params=params, **kwargs
             )
@@ -235,8 +247,9 @@ def from_params(
                 t.on_failure = on_failure
 
             tasks_all.append(t)
+            upstreams.append(_upstream)
 
-        return cls(tasks_all)
+        return cls(tasks_all), upstreams
 
     @classmethod
     def from_grid(
@@ -249,6 +262,7 @@ def from_grid(
         grid,
         name=None,
         namer=None,
+        upstream=None,
         resolve_relative_to=None,
         on_render=None,
         on_finish=None,
@@ -283,6 +297,7 @@ def from_grid(
             name=name,
             params_array=params_array,
             namer=namer,
+            upstream=upstream,
             resolve_relative_to=resolve_relative_to,
             on_render=on_render,
             on_finish=on_finish,