Method _task_from_fun

celery/app/base.py:586–629

(
        self,
        fun,
        name=None,
        base=None,
        bind=False,
        pydantic: bool = False,
        pydantic_strict: bool = False,
        pydantic_context: typing.Optional[typing.Dict[str, typing.Any]] = None,
        pydantic_dump_kwargs: typing.Optional[typing.Dict[str, typing.Any]] = None,
        **options,
    )

        return staticmethod(head_from_fun(fun, bound=bound))

    def _task_from_fun(
        self,
        fun,
        name=None,
        base=None,
        bind=False,
        pydantic: bool = False,
        pydantic_strict: bool = False,
        pydantic_context: typing.Optional[typing.Dict[str, typing.Any]] = None,
        pydantic_dump_kwargs: typing.Optional[typing.Dict[str, typing.Any]] = None,
        **options,
    ):
        if not self.finalized and not self.autofinalize:
            raise RuntimeError('Contract breach: app not finalized')
        name = name or self.gen_task_name(fun.__name__, fun.__module__)
        base = base or self.Task

        if name not in self._tasks:
            if pydantic is True:
                fun = pydantic_wrapper(self, fun, name, pydantic_strict, pydantic_context, pydantic_dump_kwargs)

            run = fun if bind else staticmethod(fun)
            task = type(fun.__name__, (base,), dict({
                'app': self,
                'name': name,
                'run': run,
                '_decorated': True,
                '__doc__': fun.__doc__,
                '__module__': fun.__module__,
                '__annotations__': _get_annotations(fun),
                '__header__': self.type_checker(fun, bound=bind),
                '__wrapped__': run}, **options))()
            # for some reason __qualname__ cannot be set in type()
            # so we have to set it here.
            try:
                task.__qualname__ = fun.__qualname__
            except AttributeError:
                pass
            self._tasks[task.name] = task
            task.bind(self)  # connects task to this app
            add_autoretry_behaviour(task, **options)
        else:
            task = self._tasks[name]
        return task

    def register_task(self, task, **options):
        """Utility for registering a task-based class.