Currently, the celery executor imports the `Task` model for the Celery database backend directly:

https://github.com/apache/airflow/blob/2dd4e96045d4a7f45cc8c06df3d25c4f1479392c/airflow/executors/celery_executor.py#L38

However, Celery, being highly customisable, uses `self.task_cls` in the backend implementation; this defaults to either `Task` or `TaskExtended` depending on the Celery configuration, see [`celery.backends.database:66`](https://github.com/celery/celery/blob/406f04a082949ac42ec7a4af94fed896c515aaa4/celery/backends/database/__init__.py#L66):

> ```python
>     task_cls = Task
> ```

and [`celery.backends.database:76,77`](https://github.com/celery/celery/blob/406f04a082949ac42ec7a4af94fed896c515aaa4/celery/backends/database/__init__.py#L76-L77):

> ```python
>         if self.extended_result:
>             self.task_cls = TaskExtended
> ```

Airflow should support this flexibility, and instead of importing a fixed class just use `backend.task_cls`, so in `_get_many_from_db_backend` just use:

```python
        Task = app.backend.task_cls
        with session_cleanup(session):
            tasks = session.query(Task).filter(Task.task_id.in_(task_ids)).all()
```

I suppose there is one downside to this: `TaskExtended` fetches a few more columns per task, but given that the result of `task.to_dict()` is passed to `backend.meta_from_decoded()`, which could also have been customised to make use of information contained in a custom model, that's neither here nor there. It would be the price anyone using the `result_extended` option would need to pay elsewhere anyway.
airflow/executors/celery_executor.py
for ti in tis:
            if ti.external_executor_id is not None:
                celery_tasks[ti.external_executor_id] = (AsyncResult(ti.external_executor_id), ti)
            else:
                not_adopted_tis.append(ti)
        if not celery_tasks:
            
            return tis
        states_by_celery_task_id = self.bulk_state_fetcher.get_many(
            map(operator.itemgetter(0), celery_tasks.values())
        )
        adopted = []
        cached_celery_backend = next(iter(celery_tasks.values()))[0].backend
        for celery_task_id, (state, info) in states_by_celery_task_id.items():
            result, ti = celery_tasks[celery_task_id]
            result.backend = cached_celery_backend
            
            
            self.adopted_task_timeouts[ti.key] = ti.queued_dttm + self.task_adoption_timeout
            self.tasks[ti.key] = result
            self.running.add(ti.key)
            self.update_task_state(ti.key, state, info)
            adopted.append(f"{ti} in state {state}")
        if adopted:
            task_instance_str = '\n\t'.join(adopted)
            self.log.info(
                "Adopted the following %d tasks from a dead executor\n\t%s", len(adopted), task_instance_str
            )
        return not_adopted_tis
def fetch_celery_task_state(async_result: AsyncResult) -> Tuple[str, Union[str, ExceptionWithTraceback], Any]: