fulltext/extract.py
def update_sent_state(sender: Optional[str] = None,
                      headers: Optional[Dict[str, str]] = None,
                      body: Any = None, **kwargs: Any) -> None:
    """Set state to SENT, so that we can tell whether a task exists."""
    celery_app = get_or_create_worker_app(current_app)
    task = celery_app.tasks.get(sender)
    backend = task.backend if task else celery_app.backend
    if headers is not None:
        backend.store_result(headers['id'], None, "SENT")
Set state to SENT, so that we can tell whether a task exists.
Set state to SENT, so that we can tell whether a task exists.
Set state to SENT so that we can tell whether a task exists .
def update_sent_state(sender: Optional[str] = None,
                      headers: Optional[Dict[str, str]] = None,
                      body: Any = None, **kwargs: Any) -> None:
    celery_app = get_or_create_worker_app(current_app)
    task = celery_app.tasks.get(sender)
    backend = task.backend if task else celery_app.backend
    if headers is not None:
        backend.store_result(headers['id'], None, "SENT")
def update_sent_state ( sender : Optional [ str ] = None , headers : Optional [ Dict [ str , str ] ] = None , body : Any = None , ** kwargs : Any ) -> None : celery_app = get_or_create_worker_app ( current_app ) task = celery_app . tasks . get ( sender ) backend = task . backend if task else celery_app . backend if headers is not None : backend . store_result ( headers [ 'id' ] , None , "SENT" )
Set state to SENT, so that we can tell whether a task exists.
Set state to SENT so that we can tell whether a task exists .
"""Set state to SENT, so that we can tell whether a task exists."""
sender
Optional[str]
headers
Optional[Dict[str, str]]
body
Any
kwargs
Any