sbdchd / celery-types Goto Github PK

View Code? Open in Web Editor NEW

66.0 4.0 35.0 208 KB

:seedling: Type stubs for Celery and its related packages

License: Apache License 2.0

Shell 0.57% Python 99.43%

celery mypy types stubs kombu billiard django-celery-results amqp

celery-types's Introduction

celery-types

Type stubs for celery related projects:

install

pip install celery-types

You'll also need to monkey patch Task so generic params can be provided:

from celery.app.task import Task
Task.__class_getitem__ = classmethod(lambda cls, *args, **kwargs: cls) # type: ignore[attr-defined]

dev

initial setup

# install poetry (https://python-poetry.org/docs/)
curl -sSL https://install.python-poetry.org | python3 -
# install node
# install yarn
npm install --global yarn

# install node dependencies
yarn

regular development

poetry config virtualenvs.in-project true
poetry install

# run formatting, linting, and typechecking
s/lint

# build and publish
poetry publish --build

celery-types's People

Stargazers

Watchers

celery-types's Issues

More typings for celery.worker.Request

Missing is_eager, called_directly, and timelimit among other things.

https://docs.celeryq.dev/en/stable/userguide/tasks.html#task-request

Change `apply_async`'s `eta` to be `Optional[datetime]`

Please change apply_async's eta to be Optional[datetime]

celery-types/typings/celery/canvas.pyi

Line 43 in 737a5ba

eta: datetime = ...,

No overload variant of "shared_task" matches argument

Hello,

I am facing this issue that I am not quite sure how to resolve
I have a simple function defined like:

@shared_task
def foo(*args:Any, **kwargs:Any)->None:
    ...
    return None

and I am getting the following error when running mypy on it.

tasks.py:86: error: No overload variant of "shared_task" matches argument type "Callable[[VarArg(Any), KwArg(Any)], None]"
tasks.py:86: note: Possible overload variants:
tasks.py:86: note:     def shared_task(*, name: str = ..., serializer: str = ..., bind: bool = ..., autoretry_for: Tuple[Type[Exception], ...] = ..., max_retries: int = ..., default_retry_delay: int = ..., acks_late: bool = ..., ignore_result: bool = ..., soft_time_limit: int = ..., time_limit: int = ..., base: None = ..., retry_kwargs: Dict[str, Any] = ..., retry_backoff: bool = ..., retry_backoff_max: int = ..., retry_jitter: bool = ..., typing: bool = ..., rate_limit: Optional[str] = ..., trail: bool = ..., send_events: bool = ..., store_errors_even_if_ignored: bool = ..., autoregister: bool = ..., track_started: bool = ..., acks_on_failure_or_timeout: bool = ..., reject_on_worker_lost: bool = ..., throws: Tuple[Type[Exception], ...] = ..., expires: Union[float, datetime, None] = ..., priority: Optional[int] = ..., resultrepr_maxsize: int = ..., request_stack: _LocalStack = ..., abstract: bool = ...) -> Callable[[Callable[..., Any]], Task]
tasks.py:86: note:     def [_T <: Task] shared_task(*, name: str = ..., serializer: str = ..., bind: bool = ..., autoretry_for: Tuple[Type[Exception], ...] = ..., max_retries: int = ..., default_retry_delay: int = ..., acks_late: bool = ..., ignore_result: bool = ..., soft_time_limit: int = ..., time_limit: int = ..., base: Type[_T], retry_kwargs: Dict[str, Any] = ..., retry_backoff: bool = ..., retry_backoff_max: int = ..., retry_jitter: bool = ..., typing: bool = ..., rate_limit: Optional[str] = ..., trail: bool = ..., send_events: bool = ..., store_errors_even_if_ignored: bool = ..., autoregister: bool = ..., track_started: bool = ..., acks_on_failure_or_timeout: bool = ..., reject_on_worker_lost: bool = ..., throws: Tuple[Type[Exception], ...] = ..., expires: Union[float, datetime, None] = ..., priority: Optional[int] = ..., resultrepr_maxsize: int = ..., request_stack: _LocalStack = ..., abstract: bool = ...) -> Callable[[Callable[..., Any]], _T]

I am actually using args and kwargs, so removing those won't solve my issue.
Thanks in advance for help,

Loic

TaskRegistry has no type information for the register() method

I define a couple task classes, then attempt to register them in my tasks.py file (on app finalize):

registry: TaskRegistry = celery_app.tasks
registry.register(PullTaskRunner)
registry.register(PushTaskRunner)

However the type information seems to be missing the register method:

run run-test: commands[1] | mypy .
...
parameters/tasks.py:21: error: "TaskRegistry" has no attribute "register"
parameters/tasks.py:22: error: "TaskRegistry" has no attribute "register"

There is a method for it though:

class TaskRegistry(dict):
    """Map of registered tasks."""
    ...
    def register(self, task):
        """Register a task in the task registry.

        The task will be automatically instantiated if not already an
        instance. Name must be configured prior to registration.
        """
        ...

Task.max_retries has wrong type

https://github.com/sbdchd/celery-types/blob/main/celery-stubs/app/task.pyi#L36
https://docs.celeryq.dev/en/v5.2.7/userguide/tasks.html#Task.max_retries

According to celery docs, it can have value None, but here it's declared as int. It should be Optional[int],

Add examples

Hi,

It would be great with some examples for both server and client side code (client with no access to server implementation, i.e. defined via celery.signature).

Currently I'm doing something like this client side

from celery import Signature, Celery

class WorkerClient(object):
    def __init__(self):
        self.app = Celery()

    def handle_request(self, a: int, b: int, task_id=None):
        sig: Signature[int] = self.app.signature("handle_request")
        options = {}
        if task_id:
            options["task_id"] = task_id
        return sig.apply_async(kwargs=dict(a=a, b=b), **options)

But it's not that typesafe on inputs (just settings kwargs). Preferably I'd define the signature with something like this where both input and return type are given.

HandleRequest = Task[[int, int], int]

sig: Sig[HandleRequest] = self.app.signature("handle_request")

Not sure if it's possible or if there are any good indicators on how to type client side code

mypy: "param_name" to "delay" of "Task" has incompatible type str; expected <nothing>

Updating from 0.15 to 0.18 (via a bandit bump to 1.7.5) seems to have broken mypy's ability to distinguish task parameters correctly. In lieu of rolling back my celery-types version I was curious if anyone knows a workaround or if I am maybe doing something incorrectly?

mypy version: 0.981

Celery.autodiscover_tasks packages param has incorrect type

Celery.autodiscover_tasks parameter packages is typed as packages: Optional[List[str]] but the implementation also allows it to be a Callable (docs), and mypy complains when a Callable is passed in
error: Argument 1 to "autodiscover_tasks" of "Celery" has incompatible type "Callable[[], List[str]]"; expected "Optional[List[str]]" [arg-type]

Argument "args" to "send_task" of "Celery" has incompatible type "Tuple[str, str, str]"; expected "Optional[Tuple[Any]]"

Hello,
i have tasks with multiple args but it seems that the type hints only accepts a single argument

Argument "args" to "send_task" of "Celery" has incompatible type "Tuple[str, str, str]"; expected "Optional[Tuple[Any]]"

the provided type annotation is:

    def send_task(
        self,
        name: str,
        args: Optional[Tuple[Any]] = ...,
[...]

maybe that it should be

        args: Optional[Tuple[Any, ...]] = ...,

to allow for args of any length?

error: "Celery" has no attribute "on_after_configure"

Typechecking for this example doesn't work:

from celery import Celery
from celery.schedules import crontab

app = Celery()

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    # Calls test('hello') every 10 seconds.
    sender.add_periodic_task(10.0, test.s('hello'), name='add every 10')

    # Calls test('world') every 30 seconds
    sender.add_periodic_task(30.0, test.s('world'), expires=10)

    # Executes every Monday morning at 7:30 a.m.
    sender.add_periodic_task(
        crontab(hour=7, minute=30, day_of_week=1),
        test.s('Happy Mondays!'),
    )

@app.task
def test(arg):
    print(arg)

@app.task
def add(x, y):
    z = x + y
    print(z)

error: "Celery" has no attribute "on_after_configure"

Probably the other signals don't work as well

"Request" has no attribute "retries"

I believe the Request object does have a retries attribute: https://github.com/celery/celery/blob/b0cfa0d818743262a032c541cce2fa8c43fabad4/celery/app/task.py#L558

But the stubs seem to be missing that. Can that be added? If there's agreement, I can try to open a PR.

`Task.request` is not `celery.worker.request.Request` (during `task_prerun`)

Task.request is typed like below

celery-types/celery-stubs/app/task.pyi

Lines 235 to 236 in af89436

    
           @property 
        
           def request(self) -> Request: ...

With the following task_prerun receiver

@signals.task_prerun.connect
def task_prerun_receiver(*args: Any, **kwargs: Any) -> None:
    task: Task = kwargs["sender"]
    ...
    logger.info("task_prerun_receiver", task_request_dict=task.request.request_dict)

I'm getting at traceback like

Traceback (most recent call last):
  File "/app/venv/lib/python3.10/site-packages/celery/utils/dispatch/signal.py", line 276, in send
    response = receiver(signal=self, sender=sender, **named)
  File "/app/src/myapp/celery/app.py", line 59, in task_prerun_receiver
    logger.info("task_prerun_receiver", task_request_dict=task.request.request_dict)
AttributeError: 'Context' object has no attribute 'request_dict'

Celery documents "Sender is the task object being executed"
Ref: https://docs.celeryq.dev/en/stable/userguide/signals.html#task-prerun

I thought task.request was a Request object, but at least in the case of a receiver on task_prerun it's rather an instance of celery.app.task.Context.

Signature apply and apply_async are incorrect

I may be wrong, but I'm under the impression that Signature.apply and Signature.apply_async are wrongly typed.

celery-types/celery-stubs/canvas.pyi

Line 67 in db87d44

args: Optional[Tuple[Any]] = ...,

celery-types/celery-stubs/canvas.pyi

Line 73 in db87d44

args: Optional[Tuple[Any]] = ...,

They should be

-       args: Optional[Tuple[Any]] = ...,
+       args: Optional[Tuple[Any, ...]] = ...,

Otherwise, we could have only one positional arg. Moreover, the other args in this file are correctly typed.

I'll open a PR soon.

Signature for group method may be incorrect?

I am creating a group using a generator of task signatures:

    job = group(
        resolve_value.s(value.id, as_of, http_meta)
        for value, as_of in work.items()
    )

This is valid according to the celery source comments:

https://github.com/celery/celery/blob/master/celery/canvas.py#L1009-L1012

Note:
    If only one argument is passed, and that argument is an iterable
    then that'll be used as the list of tasks instead: this
    allows us to use ``group`` with generator expressions.

However mypy is complaining:

value_query.py:150: error: Argument 1 to "group" has incompatible type "Generator[Signature, None, None]"; expected "Signature"

All my tests pass however, so it appears the type declaration is not complete here.
Here's how Visual Code sees it:

Module 'celery.result' has no attribute 'allow_join_result'

mypy complains even though these are valid in celery

New release?

Hi @sbdchd, any chance we could get a release out within the next few weeks?

It would be great to have the latest fixes in a released version. Let me know if there is anything I can do to help.

task_cls

Hi there. I'm working with a repository where we've overridden the default task class and added some methods, and we're using task_cls to initialize Celery with this as the default class. I'm seeing lots of errors about these custom methods not existing, and I think this is because celery-types doesn't support task_cls.

Is this something you'd be open to supporting in the future?

typehint for `celery.app.Context.parent_id` incorrect

Maybe I'm wrong, but I think Context.parent_id should be of type str | None instead of

celery-types/celery-stubs/app/task.pyi

Line 49 in 6d55b8c

parent_id: int | None

All other id-like fields are str | None and after a quick search, I found several places in celery, where str-variables are given as a named parameter called parent_id, like in celery.app.trace.build_tracer in the subfunction trace_task line 520.

I will create a PR soon.

Rename celery-types to types-celery?

Should this project be renamed given the Modular Typeshed paradigm? http://mypy-lang.blogspot.com/2021/05/the-upcoming-switch-to-modular-typeshed.html

Fix poetry build setup

Currently using the latest poetry breaks the build of the wheel

❯ tree celery_types-0.8.2-py3-none-any        
celery_types-0.8.2-py3-none-any
├── amqp-stubs
│   ├── __init__.pyi
│   └── exceptions.pyi
├── billiard-stubs
│   ├── __init__.pyi
│   ├── einfo.pyi
│   └── exceptions.pyi
├── celery-stubs
│   ├── __init__.pyi
│   ├── _state.pyi
│   ├── app
│   │   ├── __init__.pyi
│   │   ├── amqp.pyi
│   │   ├── base.pyi
│   │   ├── beat.pyi
│   │   ├── control.pyi
│   │   ├── events.pyi
│   │   ├── log.pyi
│   │   ├── registry.pyi
│   │   ├── routes.pyi
│   │   ├── task.pyi
│   │   └── utils.pyi
│   ├── apps
│   │   ├── __init__.pyi
│   │   └── worker.pyi
│   ├── backends
│   │   ├── __init__.pyi
│   │   └── base.pyi
│   ├── canvas.pyi
│   ├── concurrency
│   │   ├── __init__.pyi
│   │   └── base.pyi
│   ├── exceptions.pyi
│   ├── loaders
│   │   ├── __init__.pyi
│   │   └── base.pyi
│   ├── local.pyi
│   ├── result.pyi
│   ├── schedules.pyi
│   ├── signals.pyi
│   ├── states.pyi
│   ├── utils
│   │   ├── __init__.pyi
│   │   ├── abstract.pyi
│   │   ├── collections.pyi
│   │   ├── dispatch
│   │   │   ├── __init__.pyi
│   │   │   └── signal.pyi
│   │   ├── graph.pyi
│   │   ├── log.pyi
│   │   ├── objects.pyi
│   │   ├── term.pyi
│   │   ├── threads.pyi
│   │   └── time.pyi
│   └── worker
│       ├── __init__.pyi
│       ├── request.pyi
│       └── worker.pyi
├── celery_types-0.8.2.dist-info
│   ├── LICENSE
│   ├── METADATA
│   ├── RECORD
│   └── WHEEL
├── django_celery_results-stubs
│   ├── __init__.pyi
│   ├── managers.pyi
│   └── models.pyi
├── ephem-stubs
│   └── __init__.pyi
├── kombu-stubs
│   ├── __init__.pyi
│   ├── abstract.pyi
│   ├── connection.pyi
│   ├── entity.pyi
│   ├── exceptions.pyi
│   ├── message.pyi
│   ├── messaging.pyi
│   ├── pidbox.pyi
│   ├── pools.pyi
│   ├── resource.pyi
│   ├── transport
│   │   ├── __init__.pyi
│   │   └── base.pyi
│   └── utils
│       ├── __init__.pyi
│       └── uuid.pyi
└── vine-stubs
    ├── __init__.pyi
    ├── abstract.pyi
    ├── funtools.pyi
    ├── promises.pyi
    └── synchronization.pyi

18 directories, 74 files
❯ tree celery_types-0.8.3-py3-none-any
celery_types-0.8.3-py3-none-any
├── celery_types-0.8.3.dist-info
│   ├── LICENSE
│   ├── METADATA
│   ├── RECORD
│   └── WHEEL
└── typings
    ├── amqp
    │   ├── __init__.pyi
    │   └── exceptions.pyi
    ├── billiard
    │   ├── __init__.pyi
    │   ├── einfo.pyi
    │   └── exceptions.pyi
    ├── celery
    │   ├── __init__.pyi
    │   ├── _state.pyi
    │   ├── app
    │   │   ├── __init__.pyi
    │   │   ├── amqp.pyi
    │   │   ├── base.pyi
    │   │   ├── beat.pyi
    │   │   ├── control.pyi
    │   │   ├── events.pyi
    │   │   ├── log.pyi
    │   │   ├── registry.pyi
    │   │   ├── routes.pyi
    │   │   ├── task.pyi
    │   │   └── utils.pyi
    │   ├── apps
    │   │   ├── __init__.pyi
    │   │   └── worker.pyi
    │   ├── backends
    │   │   ├── __init__.pyi
    │   │   └── base.pyi
    │   ├── canvas.pyi
    │   ├── concurrency
    │   │   ├── __init__.pyi
    │   │   └── base.pyi
    │   ├── exceptions.pyi
    │   ├── loaders
    │   │   ├── __init__.pyi
    │   │   └── base.pyi
    │   ├── local.pyi
    │   ├── result.pyi
    │   ├── schedules.pyi
    │   ├── signals.pyi
    │   ├── states.pyi
    │   ├── utils
    │   │   ├── __init__.pyi
    │   │   ├── abstract.pyi
    │   │   ├── collections.pyi
    │   │   ├── dispatch
    │   │   │   ├── __init__.pyi
    │   │   │   └── signal.pyi
    │   │   ├── graph.pyi
    │   │   ├── log.pyi
    │   │   ├── objects.pyi
    │   │   ├── term.pyi
    │   │   ├── threads.pyi
    │   │   └── time.pyi
    │   └── worker
    │       ├── __init__.pyi
    │       ├── request.pyi
    │       └── worker.pyi
    ├── django_celery_results
    │   ├── __init__.pyi
    │   ├── managers.pyi
    │   └── models.pyi
    ├── ephem
    │   └── __init__.pyi
    ├── kombu
    │   ├── __init__.pyi
    │   ├── abstract.pyi
    │   ├── connection.pyi
    │   ├── entity.pyi
    │   ├── exceptions.pyi
    │   ├── message.pyi
    │   ├── messaging.pyi
    │   ├── pidbox.pyi
    │   ├── pools.pyi
    │   ├── resource.pyi
    │   ├── transport
    │   │   ├── __init__.pyi
    │   │   └── base.pyi
    │   └── utils
    │       ├── __init__.pyi
    │       └── uuid.pyi
    └── vine
        ├── __init__.pyi
        ├── abstract.pyi
        ├── funtools.pyi
        ├── promises.pyi
        └── synchronization.pyi

19 directories, 74 files

0.8.3 regression

Hello,
just after the release of version 0.8.3 my nightly mypy CI build failed due to errors in celery typing

Once pinned celery-types to 0.8.2 everything back working

May it be some regression introduced in 0.8.3? It is seems that with the new version celery is no longer recognized as typed

error: Skipping analyzing "celery.result": found module but no type hints or library stubs
error: Skipping analyzing "celery.app.task": found module but no type hints or library stubs
error: Argument 1 to "xyz" becomes "Any" due to an unfollowed import
error: Skipping analyzing "celery": found module but no type hints or library stubs
error: Skipping analyzing "celery.exceptions": found module but no type hints or library stubs

Wrong type for Inspect class

Hi there,

Maybe it's the documentation of Celery that's unclear but for celery.app.control.Inspect the method stats say Ref:

Returns: Dictionary {HOSTNAME: STAT_INFO}.
Return type: Dict

But inside your code I see:

celery-types/celery-stubs/app/control.pyi

Line 33 in 0c29be1

def stats(self) -> list[dict[str, _Reply]]: ...

I have the same issue with query_task.

If I follow the rabbit:

Inspect.stats(): return self._request('stats')
Inspect._request(...): return self._prepare(...)
And Inspect._prepare(...):

    def _prepare(self, reply):
        if reply:
            # flatten_reply convert [{"HOSTNAME": Reply}, {"HOSTNAME": Reply}, ...]
            # to {"HOSTNAME": Reply, "HOSTNAME": Reply, ...}
            # so list[dict[str, _Reply]] to dict[str, _Reply]
            by_node = flatten_reply(reply)
            if (self.destination and
                    not isinstance(self.destination, (list, tuple))):
                return by_node.get(self.destination)
            if self.pattern:
                pattern = self.pattern
                matcher = self.matcher
                # Return a dict[str, _Reply]
                return {node: reply for node, reply in by_node.items()
                        if match(node, pattern, matcher)}
            return by_node

Ref for Inspect._prepare

I'm pretty new with django, celery, etc (about a week old) so maybe I just miss something, thanks if someone can explain it to me it'll be more than welcome.

Incorrect typing for autoretry_for

We have the following code

@shared_task(
    autoretry_for=(PoolTimeout, CancelledError),
    max_retries=10,
    retry_backoff=True,
    retry_backoff_max=100,
    retry_jitter=False,
)
def my_task(...):
  ...

But unfortunately mypy it's throwing the following error:

error: Argument "autoretry_for" to "shared_task" has incompatible type "Tuple[Type[PoolTimeout], Type[CancelledError]]"; expected "Tuple[Type[Exception], ...]"  [arg-type]

The reason for this is that CancelledError extends from BaseException instead of Exception. This is the issue where they changed the base class.

You can't use app.control.broadcast() without mypy complaining

The following code leads to a mypy complaint (broadcast.py:5: error: "Control" has no attribute "broadcast"):

from celery import Celery

app = Celery()

app.control.broadcast()

broadcast() is a real method of Control, which you've defined as ...

it'd be cool if Control could be made less strict while these annotations don't exist

TypeError: 'type' object is not subscriptable

Hello!

I have a problem with the new Task annotations.
I installed version 0.13.0 and started receiving typing errors on Task annotations (from #75)

Missing type parameters for generic type "Task"

So added the expected annotations, like:

from celery.app.task import Task
from typing import Any

@app.task(bind=True)
def my_task(self: Task[Any, Any]):

But now celery (version 5.2.7) fails at runtime with:

    self: Task[Any, Any],
          │    │    └ typing.Any
          │    └ typing.Any
          └ <class 'celery.app.task.Task'>

TypeError: 'type' object is not subscriptable

What I'm missing? 😢

0.15.0 -> 0.17.0 error: "List[Dict[str, Any]]" has no attribute "items" [attr-defined]

Hi! Thanks for creating celery-types.

Since 0.17.0, the following type seems to be wrongly identified. The code below shows a working example. I'm guessing that inspect.active() just returns a dict.

def example(inspect: celery.app.control.Inspect):
    active = inspect.active()
    for _, tasks in active.items():   # error: List[Dict[str, Any]]" has no attribute "items"  [attr-defined] ## happens here
        print(task["type"])

Add support for DjangoTask's delay_on_commit and apply_async_on_commit methods

Celery 5.4 introduced some shortcuts for Django projects using a custom task class. This new DjangoTask base class adds the following methods:

delay_on_commit
apply_async_on_commit

https://docs.celeryq.dev/en/stable/django/first-steps-with-django.html#trigger-tasks-at-the-end-of-the-database-transaction

Not sure what's the best way to support app.task/shared_task returning instances of this base class but we can start by adding stubs for celery.contrib.django.task.DjangoTask (see API reference).

celery.app.Task.throws is a tuple of exception types?

The type seems to be incorrect?

celery-types/celery-stubs/app/task.pyi

Line 35 in 4ac7923

throws: Tuple[Exception, ...]

In the other locations it is expressed as a tuple of exception types:

https://github.com/sbdchd/celery-types/search?q=throws

Split the package into small packages

👋 Hello, I am interested in the kombu stubs and nothing else.

From what I saw, there is no way to simply use the kombu typings, I need to download everything.

What are your thoughts on publishing each package individually and making the current repository an aggregation of the individual parts?

Missing types for `celery.app.log.TaskFormatter`

Right now, there are no stubs for celery.app.log.TaskFormatter, which raises a mypy error

wrong type for `shared_task`'s base param

shared_task should be updated to match the other task method

mypy: "<some_param> to "delay" of "Task" has incompatible type <some_type>; expected <nothing>

error: "Callable[[Task], Request]" has no attribute "id"

Hello,
with bound tasks it is possible to access the task ID with self.request.id, as documented in:

https://docs.celeryproject.org/en/stable/userguide/tasks.html#bound-tasks

logger = get_task_logger(__name__)

@app.task(bind=True)
def add(self, x, y):
    logger.info(self.request.id)

But it is not recognized and mypy raises error: "Callable[[Task], Request]" has no attribute "id"

Thank you very much for your work!

Mark stubs as `partial` per PEP-561

Please, consider marking these stubs as partial per PEP-561. Otherwise, Pylance, for example, reports all untyped imports as missing (try from celery.backends.database.models import TaskExtended).

Why not typeshed?

The typeshed repository is certainly the right place where to put those types. Why not submitting the project to them?

Any chance the version numbers here could track the version numbers in Celery?

Using Celery 5.2.7 but
celery-types version 0.14.0

Maybe it's just not there yet....

Thank you 🙂

celery-types/typings/celery/app/__init__.pyi

Line 13 in e1750be

def shared_task(

The args in this function declaration helped me catch an issue. I was using:

@celery.shared_task(
  exponential_backoff = 5
)
def some_func(...):

I thought exponential_backoff was a celery decorator param... turns out it's not! This python package is very helpful because now my type linter caught that issue. Thanks!

[Question] self annotation

How should I annotate celery when bind=True with self?

Solved by below answer by @sbdchd.

you can use type alias:

from typing import Any

from celery import Task

SelfTask = Task[Any, Any]


@app.task(bind=True)
def add_2(self: SelfTask, x: int, y: int) -> None:
    logger.info(self.request.id)

tags so other can find it:

celery type annotation task self

Imports from celery.contrib could not be resolved

When importing stuff from celery.contrib (e.g. abortable.AbortableTask) pyright tells me it could not be resolved. It seems to be missing in the stubs.

Request: add 'dont_autoretry_for' to task signature

Since celery 5.3.0 the dont_autoretry_for parameter was added to the task parameters: https://docs.celeryq.dev/en/stable/userguide/tasks.html#Task.dont_autoretry_for

This is currently not supported by the celery-types package, is it possible to add this signature?

celery.signature should accept str

first of all, thank you for creating this package!

I have an issue with celery.signature not accepting str:

error: Argument 1 to "signature" has incompatible type "str"; expected "Union[Signature, Dict[str, Any]]"  [arg-type]

The celery documentation shows an example with str, https://docs.celeryproject.org/en/stable/userguide/canvas.html
Which contradicts the source code comments here: https://github.com/celery/celery/blob/5d72aeedb6329b609469c63998e9335e017bd204/celery/canvas.py#L1568

You can see that Signature which is called by signature supports str here: https://github.com/celery/celery/blob/5d72aeedb6329b609469c63998e9335e017bd204/celery/canvas.py#L149

Make more params in `task()` optional?

I believe most of the args in task() should be optional, eg. all the retry params, and acks_late, among others?

celery-types/celery-stubs/app/base.pyi

Line 110 in 7165479

def task(

Group is missing tasks

I noticed that canvas.group is using self.tasks a bit everywhere, but it's never declared in any __init__ (I might have missed it though).
And the stubs don't declare it either. That has the effect of triggering mypy when doing mygroup.tasks.

error: "group" has no attribute "tasks"  [attr-defined]

I'm not sure if this is supposed to be a stub-only fix (I could do it) or if it highlights a deeper potential bug, where we could have branches that would raise AttributeError (I'll need some help, and a ticket would be opened on celery repo).

Release schedule

Is there a release schedule/process for this project? When will merged branches be released to pypi?

Task inheritance typechecking is broken

https://docs.celeryproject.org/en/stable/userguide/tasks.html?highlight=inheritance#task-inheritance

import celery

app = celery.Celery("worker")


class MyTask(celery.Task):

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        print('{0!r} failed: {1!r}'.format(task_id, exc))

@app.task(base=MyTask)
def add(x, y):
    raise KeyError()

test.py:11: error: Argument "base" to "task" of "Celery" has incompatible type "Type[MyTask]"; expected "Optional[Task]"

Add types for `kombu.utils.json`

Hello,

Thanks for this collection of type stubs!

I've been using it in a project, and recently i've had to use kombu JSON serializers, and mypy is reporting error: Module "kombu.utils" has no attribute "json" [attr-defined] because the types for that module are missing.

Calling a chord object should return an AsyncResult

When you "call" a chord object, it delegates to apply_async, so the return value of something like

chord([task1.s(), task2.s()])(callback.s())

should be AsyncResult

I tried adding this:

    def __call__(
        self,
        body: Optional[Signature[Any]] = ...,
        **options: Any,
    ) -> celery.result.AsyncResult[Any]: ...

but mypy complains because this disagrees with Signature's typing:

typings/celery/canvas.pyi:376:5: error: Signature of "__call__" incompatible with supertype "Signature"  [override]
typings/celery/canvas.pyi:376:5: note:      Superclass:
typings/celery/canvas.pyi:376:5: note:          def __call__(self, *partial_args: Any, **partial_kwargs: Any) -> Any
typings/celery/canvas.pyi:376:5: note:      Subclass:
typings/celery/canvas.pyi:376:5: note:          def __call__(self, body: Optional[Signature[Any]] = ..., **options: Any) -> AsyncResult[Any]

is it possible to type this correctly?

sbdchd / celery-types Goto Github PK

celery-types's Introduction

celery-types

install

dev

initial setup

regular development

related

celery-types's People

Stargazers

Watchers

Forkers

celery-types's Issues

Recommend Projects

Recommend Topics

Recommend Org