API Reference¶

@runtime_checkable
class CeleryClient(Protocol):
    """Marker protocol for Celery client classes.

    Classes decorated with ``@celery`` may optionally implement this
    protocol. It carries no methods -- its sole purpose is to provide a
    type that can be used for ``isinstance`` checks and type-hint
    annotations.
    """

    pass

@component
class CeleryClientInterceptor(MethodInterceptor):
    """AOP interceptor that converts method calls into ``send_task`` invocations.

    Registered as a pico-ioc ``@component`` and injected with the
    ``Celery`` application. When a ``@send_task``-decorated method is
    called, this interceptor reads the stored metadata and delegates to
    ``celery_app.send_task()``.

    Args:
        celery_app: The ``Celery`` application instance provided by
            ``CeleryFactory``.
    """

    def __init__(self, celery_app: Celery):
        self._celery = celery_app

    def invoke(self, ctx: MethodCtx, call_next: Callable[[MethodCtx], Any]) -> Any:
        """Intercept a method call and dispatch it as a Celery task.

        If the called method carries ``@send_task`` metadata, the
        interceptor extracts the task name and options and calls
        ``celery_app.send_task()``. Otherwise the call is forwarded to
        the next handler in the interceptor chain.

        Args:
            ctx: The method invocation context containing the class,
                method name, positional args, and keyword args.
            call_next: Callable to invoke the next interceptor or the
                original method.

        Returns:
            A ``celery.result.AsyncResult`` when the method is a
            ``@send_task`` sender, or the result of *call_next* for
            non-sender methods.
        """
        try:
            original_func = getattr(ctx.cls, ctx.name)
            meta = getattr(original_func, PICO_CELERY_SENDER_META, None)
        except AttributeError:
            meta = None

        if not meta:
            return call_next(ctx)

        task_name = meta["name"]
        options = meta.get("options", {})

        return self._celery.send_task(task_name, args=ctx.args, kwargs=ctx.kwargs, **options)

@configured(target="self", prefix="celery", mapping="tree")
@dataclass
class CelerySettings:
    """Settings required to create a Celery application instance.

    Fields are populated from the pico-ioc configuration source under
    the ``"celery"`` prefix. For example, a ``DictSource`` entry of
    ``{"celery": {"broker_url": "redis://..."}}`` maps to
    ``CelerySettings.broker_url``.

    Attributes:
        broker_url: URL of the message broker
            (e.g. ``"redis://localhost:6379/0"``).
        backend_url: URL of the result backend
            (e.g. ``"redis://localhost:6379/1"``).
        task_track_started: When ``True``, Celery reports a ``STARTED``
            state when a worker begins executing a task. Defaults to
            ``True``.

    Example:
        .. code-block:: python

            from pico_ioc import configuration, DictSource

            config = configuration(DictSource({
                "celery": {
                    "broker_url": "redis://localhost:6379/0",
                    "backend_url": "redis://localhost:6379/1",
                }
            }))
    """

    broker_url: str
    backend_url: str

    task_track_started: bool = True

@factory
class CeleryFactory:
    """IoC factory that provides a singleton ``Celery`` application.

    Registered with pico-ioc via the ``@factory`` decorator. Its
    ``create_celery_app`` method is discovered automatically and called
    once to produce the shared ``Celery`` instance.

    Example:
        The factory is auto-discovered; no manual instantiation is needed.
        Simply inject ``Celery`` wherever required:

        .. code-block:: python

            from celery import Celery
            from pico_ioc import component

            @component
            class MyService:
                def __init__(self, app: Celery):
                    self._app = app
    """

    @provides(Celery, scope="singleton")
    def create_celery_app(self, settings: CelerySettings) -> Celery:
        """Create and configure a ``Celery`` application instance.

        Args:
            settings: A ``CelerySettings`` instance populated from the
                pico-ioc configuration tree.

        Returns:
            A fully configured ``Celery`` application with the broker,
            backend, and tracking options applied.
        """
        celery_app = Celery(
            "pico_celery_tasks",
            broker=settings.broker_url,
            backend=settings.backend_url,
        )
        celery_app.conf.update(
            task_track_started=settings.task_track_started,
        )
        return celery_app

@component
class PicoTaskRegistrar:
    """Discovers ``@task`` methods and registers them with the Celery app.

    This component is created during container initialisation. Its
    ``register_tasks`` method (annotated with ``@configure``) is called
    automatically by pico-ioc after all components are registered,
    ensuring that every ``@task``-decorated async method is made available
    to the Celery worker.

    Args:
        container: The pico-ioc ``PicoContainer`` used for component
            resolution at task execution time.
        celery_app: The ``Celery`` application instance (provided by
            ``CeleryFactory``) on which tasks are registered.
    """

    def __init__(self, container: PicoContainer, celery_app: Celery):
        self._container = container
        self._celery_app = celery_app

    @configure
    def register_tasks(self) -> None:
        """Scan all container components and register ``@task`` methods.

        Iterates over every component registered in the container's
        locator. For each class that has methods decorated with
        ``@task``, a synchronous wrapper is created and registered on the
        Celery application via ``celery_app.task()``.

        This method is invoked automatically by pico-ioc during the
        ``@configure`` lifecycle phase.
        """
        locator = getattr(self._container, "_locator", None)
        if locator is None:
            return
        metadata_map = getattr(locator, "_metadata", {})
        for md in metadata_map.values():
            component_cls = getattr(md, "concrete_class", None)
            if not inspect.isclass(component_cls):
                continue
            for method_name, method_func in inspect.getmembers(component_cls, inspect.isfunction):
                if not hasattr(method_func, PICO_CELERY_METHOD_META):
                    continue
                meta = getattr(method_func, PICO_CELERY_METHOD_META)
                task_name = meta.get("name")
                celery_options = meta.get("options", {})
                wrapper = self._create_task_wrapper(component_cls, method_name, self._container)
                self._celery_app.task(name=task_name, **celery_options)(wrapper)

    def _create_task_wrapper(
        self, component_cls: Type, method_name: str, container: PicoContainer
    ) -> Callable[..., Any]:
        """Build a sync wrapper that resolves the component and runs the async task.

        The wrapper handles event-loop detection: if no loop is running it
        uses ``asyncio.run()``; if a loop is already active (e.g. inside
        an ``eventlet``/``gevent`` worker) it offloads execution to a
        ``ThreadPoolExecutor`` to avoid blocking.

        Args:
            component_cls: The ``@component`` class that owns the task
                method.
            method_name: Name of the ``@task``-decorated method on
                *component_cls*.
            container: The ``PicoContainer`` used to resolve a fresh
                instance of *component_cls* for each execution.

        Returns:
            A synchronous callable suitable for registration with
            ``celery_app.task()``.
        """

        def sync_task_executor(*args: Any, **kwargs: Any) -> Any:
            async def run_task_logic() -> Any:
                component_instance = await container.aget(component_cls)
                task_method = getattr(component_instance, method_name)
                return await task_method(*args, **kwargs)

            try:
                loop = asyncio.get_running_loop()
            except RuntimeError:
                loop = None

            if loop and loop.is_running():
                import concurrent.futures

                with concurrent.futures.ThreadPoolExecutor() as pool:
                    return pool.submit(asyncio.run, run_task_logic()).result()

            return asyncio.run(run_task_logic())

        return sync_task_executor

@configured(target="self", prefix="celery", mapping="tree")
@dataclass
class CelerySettings:
    """Settings required to create a Celery application instance.

    Fields are populated from the pico-ioc configuration source under
    the ``"celery"`` prefix. For example, a ``DictSource`` entry of
    ``{"celery": {"broker_url": "redis://..."}}`` maps to
    ``CelerySettings.broker_url``.

    Attributes:
        broker_url: URL of the message broker
            (e.g. ``"redis://localhost:6379/0"``).
        backend_url: URL of the result backend
            (e.g. ``"redis://localhost:6379/1"``).
        task_track_started: When ``True``, Celery reports a ``STARTED``
            state when a worker begins executing a task. Defaults to
            ``True``.

    Example:
        .. code-block:: python

            from pico_ioc import configuration, DictSource

            config = configuration(DictSource({
                "celery": {
                    "broker_url": "redis://localhost:6379/0",
                    "backend_url": "redis://localhost:6379/1",
                }
            }))
    """

    broker_url: str
    backend_url: str

    task_track_started: bool = True

@runtime_checkable
class CeleryClient(Protocol):
    """Marker protocol for Celery client classes.

    Classes decorated with ``@celery`` may optionally implement this
    protocol. It carries no methods -- its sole purpose is to provide a
    type that can be used for ``isinstance`` checks and type-hint
    annotations.
    """

    pass

@component
class CeleryClientInterceptor(MethodInterceptor):
    """AOP interceptor that converts method calls into ``send_task`` invocations.

    Registered as a pico-ioc ``@component`` and injected with the
    ``Celery`` application. When a ``@send_task``-decorated method is
    called, this interceptor reads the stored metadata and delegates to
    ``celery_app.send_task()``.

    Args:
        celery_app: The ``Celery`` application instance provided by
            ``CeleryFactory``.
    """

    def __init__(self, celery_app: Celery):
        self._celery = celery_app

    def invoke(self, ctx: MethodCtx, call_next: Callable[[MethodCtx], Any]) -> Any:
        """Intercept a method call and dispatch it as a Celery task.

        If the called method carries ``@send_task`` metadata, the
        interceptor extracts the task name and options and calls
        ``celery_app.send_task()``. Otherwise the call is forwarded to
        the next handler in the interceptor chain.

        Args:
            ctx: The method invocation context containing the class,
                method name, positional args, and keyword args.
            call_next: Callable to invoke the next interceptor or the
                original method.

        Returns:
            A ``celery.result.AsyncResult`` when the method is a
            ``@send_task`` sender, or the result of *call_next* for
            non-sender methods.
        """
        try:
            original_func = getattr(ctx.cls, ctx.name)
            meta = getattr(original_func, PICO_CELERY_SENDER_META, None)
        except AttributeError:
            meta = None

        if not meta:
            return call_next(ctx)

        task_name = meta["name"]
        options = meta.get("options", {})

        return self._celery.send_task(task_name, args=ctx.args, kwargs=ctx.kwargs, **options)

@factory
class CeleryFactory:
    """IoC factory that provides a singleton ``Celery`` application.

    Registered with pico-ioc via the ``@factory`` decorator. Its
    ``create_celery_app`` method is discovered automatically and called
    once to produce the shared ``Celery`` instance.

    Example:
        The factory is auto-discovered; no manual instantiation is needed.
        Simply inject ``Celery`` wherever required:

        .. code-block:: python

            from celery import Celery
            from pico_ioc import component

            @component
            class MyService:
                def __init__(self, app: Celery):
                    self._app = app
    """

    @provides(Celery, scope="singleton")
    def create_celery_app(self, settings: CelerySettings) -> Celery:
        """Create and configure a ``Celery`` application instance.

        Args:
            settings: A ``CelerySettings`` instance populated from the
                pico-ioc configuration tree.

        Returns:
            A fully configured ``Celery`` application with the broker,
            backend, and tracking options applied.
        """
        celery_app = Celery(
            "pico_celery_tasks",
            broker=settings.broker_url,
            backend=settings.backend_url,
        )
        celery_app.conf.update(
            task_track_started=settings.task_track_started,
        )
        return celery_app

@component
class PicoTaskRegistrar:
    """Discovers ``@task`` methods and registers them with the Celery app.

    This component is created during container initialisation. Its
    ``register_tasks`` method (annotated with ``@configure``) is called
    automatically by pico-ioc after all components are registered,
    ensuring that every ``@task``-decorated async method is made available
    to the Celery worker.

    Args:
        container: The pico-ioc ``PicoContainer`` used for component
            resolution at task execution time.
        celery_app: The ``Celery`` application instance (provided by
            ``CeleryFactory``) on which tasks are registered.
    """

    def __init__(self, container: PicoContainer, celery_app: Celery):
        self._container = container
        self._celery_app = celery_app

    @configure
    def register_tasks(self) -> None:
        """Scan all container components and register ``@task`` methods.

        Iterates over every component registered in the container's
        locator. For each class that has methods decorated with
        ``@task``, a synchronous wrapper is created and registered on the
        Celery application via ``celery_app.task()``.

        This method is invoked automatically by pico-ioc during the
        ``@configure`` lifecycle phase.
        """
        locator = getattr(self._container, "_locator", None)
        if locator is None:
            return
        metadata_map = getattr(locator, "_metadata", {})
        for md in metadata_map.values():
            component_cls = getattr(md, "concrete_class", None)
            if not inspect.isclass(component_cls):
                continue
            for method_name, method_func in inspect.getmembers(component_cls, inspect.isfunction):
                if not hasattr(method_func, PICO_CELERY_METHOD_META):
                    continue
                meta = getattr(method_func, PICO_CELERY_METHOD_META)
                task_name = meta.get("name")
                celery_options = meta.get("options", {})
                wrapper = self._create_task_wrapper(component_cls, method_name, self._container)
                self._celery_app.task(name=task_name, **celery_options)(wrapper)

    def _create_task_wrapper(
        self, component_cls: Type, method_name: str, container: PicoContainer
    ) -> Callable[..., Any]:
        """Build a sync wrapper that resolves the component and runs the async task.

        The wrapper handles event-loop detection: if no loop is running it
        uses ``asyncio.run()``; if a loop is already active (e.g. inside
        an ``eventlet``/``gevent`` worker) it offloads execution to a
        ``ThreadPoolExecutor`` to avoid blocking.

        Args:
            component_cls: The ``@component`` class that owns the task
                method.
            method_name: Name of the ``@task``-decorated method on
                *component_cls*.
            container: The ``PicoContainer`` used to resolve a fresh
                instance of *component_cls* for each execution.

        Returns:
            A synchronous callable suitable for registration with
            ``celery_app.task()``.
        """

        def sync_task_executor(*args: Any, **kwargs: Any) -> Any:
            async def run_task_logic() -> Any:
                component_instance = await container.aget(component_cls)
                task_method = getattr(component_instance, method_name)
                return await task_method(*args, **kwargs)

            try:
                loop = asyncio.get_running_loop()
            except RuntimeError:
                loop = None

            if loop and loop.is_running():
                import concurrent.futures

                with concurrent.futures.ThreadPoolExecutor() as pool:
                    return pool.submit(asyncio.run, run_task_logic()).result()

            return asyncio.run(run_task_logic())

        return sync_task_executor