def METHOD_NAME(self, pipeline: Pipeline, *tasks: Task) -> Pipeline:
    """
    The background logic for this method is that during
    Airflow backend lineage, we compute one task at
    a time.
    Let's generalise a bit the approach by preparing
    a method capable of updating a tuple of tasks
    from the client.
    Latest changes leave all the task management
    to the client. Therefore, a Pipeline will only contain
    the tasks sent in each PUT from the client.
    """
    # Get the names of all incoming tasks
    updated_tasks_names = {task.name for task in tasks}
    # Check which tasks are currently in the pipeline but not being updated
    not_updated_tasks = []
    if pipeline.tasks:
        not_updated_tasks = [
            task for task in pipeline.tasks if task.name not in updated_tasks_names
        ]
    # All tasks are the union of the incoming tasks & the not updated tasks
    all_tasks = [*tasks, *not_updated_tasks]
    updated_pipeline = CreatePipelineRequest(
        name=pipeline.name,
        displayName=pipeline.displayName,
        description=pipeline.description,
        sourceUrl=pipeline.sourceUrl,
        concurrency=pipeline.concurrency,
        pipelineLocation=pipeline.pipelineLocation,
        startDate=pipeline.startDate,
        service=pipeline.service.fullyQualifiedName,
        tasks=all_tasks,
        owner=pipeline.owner,
        tags=pipeline.tags,
    )
    return self.create_or_update(updated_pipeline)