o wvXjA@sdZddlmZmZddlZddlmZddlmZddl m Z ddl m Z ddl mZdd lmZdd lmZdd lmZdd lmZdd lmZdZeeZdZddZddZGdddeZ GdddeZ!dS)zWorker Remote Control Client. Client for worker remote control commands. Server implementation is in :mod:`celery.worker.control`. )absolute_importunicode_literalsN) TERM_SIGNAMEmatch)Mailbox)register_after_fork)lazy)cached_property)DuplicateNodenameWarning)items) get_logger) pluralize)InspectControl flatten_replyzReceived multiple replies from node {0}: {1}. Please make sure you give each node a unique nodename using the celery worker `-n` option.c sfit|D]}fdd|D|qr1tttttdd t S)zFlatten node replies. Convert from a list of replies in this format:: [{'a@example.com': reply}, {'b@example.com': reply}] into this format:: {'a@example.com': reply, 'b@example.com': reply} csg|] }|vr|qS)add).0namedupesnodesrO/var/www/html/myproject/venv/lib/python3.10/site-packages/celery/app/control.py 1sz!flatten_reply..rz, ) setupdatewarningswarnr W_DUPNODEformatrlenjoinsorted)replyitemrrrr"s  rc CsFz|WdSty"}ztjd|ddWYd}~dSd}~ww)Nzafter fork raised exception: %r)exc_info) _after_fork Exceptionloggerinfo)controlexcrrr_after_fork_cleanup_control<s r.c@seZdZdZdZ   d.ddZddZdd Zd d Zd d Z d/ddZ d/ddZ d/ddZ ddZ ddZddZeZd/ddZddZddZd0d!d"Zd/d#d$Zd%d&Zd1d(d)Zd2d,d-ZdS)3rzAPI for app.control.inspect.N?c Cs:|p|j|_||_||_||_||_||_||_||_dSN)app destinationtimeoutcallback connectionlimitpatternmatcher) selfr2r3r4r5r1r6r7r8rrr__init__Hs  zInspect.__init__cs`|r.t|}|jrt|jttfs||jS|jr,|j|jfddt|DS|SdS)Ncs"i|] \}}t|r||qSrr)rnoder$r8r7rr ]s  z$Inspect._prepare..) rr2 isinstancelisttuplegetr7r8r )r9r$by_noderr<r_prepareTs zInspect._preparecKs6||jjj|||j|j|j|j|jd|j |j d S)NT) argumentsr2r4r5r6r3r$r7r8) rCr1r, broadcastr2r4r5r6r3r7r8)r9commandkwargsrrr_requestas zInspect._requestcC |dS)NreportrHr9rrrrJm zInspect.reportcCrI)NclockrKrLrrrrNprMz Inspect.clockcCrI)NactiverKr9saferrrrOss zInspect.activecCrI)N scheduledrKrPrrrrRyrMzInspect.scheduledcCrI)NreservedrKrPrrrrS|rMzInspect.reservedcCrI)NstatsrKrLrrrrTrMz Inspect.statscCrI)NrevokedrKrLrrrrUrMzInspect.revokedcG|jd|dS)N registered) taskinfoitemsrK)r9rXrrrrWzInspect.registeredcCrI)NpingrK)r9r2rrrrZrMz Inspect.pingcCrI)N active_queuesrKrLrrrr[rMzInspect.active_queuescGs4t|dkrt|dttfr|d}|jd|dS)Nr&r query_task)ids)r!r>r?r@rH)r9r]rrrr\szInspect.query_taskFcCrV)Nconf) with_defaultsrK)r9r_rrrr^rYz Inspect.confcCs|jd||dS)Nhello) from_noderUrK)r9rarUrrrr`sz Inspect.hellocCrI)N memsamplerKrLrrrrbrMzInspect.memsample cCrV)Nmemdump)samplesrK)r9rerrrrdrYzInspect.memdumpRequestcCs|jd|||dS)Nobjgraph)num max_depthtyperK)r9rknrjrrrrhszInspect.objgraph)Nr/NNNNNNr0)F)rc)rfrgrc)__name__ __module__ __qualname____doc__r1r:rCrHrJrNrOrRrSrTrUrWregistered_tasksrZr[r\r^r`rbrdrhrrrrrCs4           rc@seZdZdZeZd1ddZddZeddZd1d d Z e Z d2d d Z dd e fddZ de fddZd3ddZd1ddZ  d4ddZd1ddZ  d5ddZd1ddZd1d d!Zd6d#d$Zd6d%d&Zd1d'd(Zd1d)d*Z  d7d+d,Zd1d-d.Z   d8d/d0ZdS)9rzWorker remote control client.Nc sR|_j|jjddgtfdd|jj|jj|jj|jjd_tt dS)Nfanoutjsoncs jjjSr0)r1amqp producer_poolrrLrrs z"Control.__init__..)rkacceptru queue_ttlreply_queue_ttl queue_expiresreply_queue_expires) r1rr^control_exchanger control_queue_ttlcontrol_queue_expiresmailboxrr.)r9r1rrLrr:s zControl.__init__cCs |j`dSr0)rrurLrrrr(rMzControl._after_forkcCs|jjtddS)Nzcontrol.inspect)reverse)r1subclass_with_selfrrLrrrinspectszControl.inspectcCsB|j|}|jj|WdS1swYdS)aDiscard all waiting tasks. This will ignore all tasks waiting for execution, and they will be deleted from the messaging server. Arguments: connection (kombu.Connection): Optional specific connection instance to use. If not provided a connection will be acquired from the connection pool. Returns: int: the number of tasks discarded. N)r1connection_or_acquirert TaskConsumerpurge)r9r5connrrrrs$z Control.purgecCs|jd|d|||dddS)Nelection)idtopicaction)r5r2rDrE)r9rrrr5rrrrs  zControl.electionFcKs|jd||||dd|S)a[Tell all (or specific) workers to revoke a task by id (or list of ids). If a task is revoked, the workers will ignore the task and not execute it after all. Arguments: task_id (Union(str, list)): Id of the task to revoke (or list of ids). terminate (bool): Also terminate the process currently working on the task (if any). signal (str): Name of signal to send to process if terminate. Default is TERM. See Also: :meth:`broadcast` for supported keyword arguments. revoke)task_id terminatesignalr2rDNrr)r9rr2rrrGrrrrszControl.revokecKs|j|f|d|d|S)zTell all (or specific) workers to terminate a task by id (or list of ids). See Also: This is just a shortcut to :meth:`revoke` with the terminate argument enabled. T)r2rrr)r9rr2rrGrrrrszControl.terminater/cKs|j ddi||d|S)zPing all (or specific) workers. Returns: List[Dict]: List of ``{'hostname': reply}`` dictionaries. See Also: :meth:`broadcast` for supported keyword arguments. rZT)r$rDr2r3N)rZr)r9r2r3rGrrrrZs z Control.pingcKs|j d|||dd|S)aTell workers to set a new rate limit for task by type. Arguments: task_name (str): Name of task to change rate limit for. rate_limit (int, str): The rate limit as tasks per second, or a rate limit string (`'100/m'`, etc. see :attr:`celery.task.base.Task.rate_limit` for more information). See Also: :meth:`broadcast` for supported keyword arguments. rate_limit) task_namerrN)rr)r9rrr2rGrrrr s zControl.rate_limitdirectc Ks2|j d|t||||dfi|pid|S)aTell all (or specific) workers to start consuming from a new queue. Only the queue name is required as if only the queue is specified then the exchange/routing key will be set to the same name ( like automatic queues do). Note: This command does not respect the default queue/exchange options in the configuration. Arguments: queue (str): Name of queue to start consuming from. exchange (str): Optional name of exchange. exchange_type (str): Type of exchange (defaults to 'direct') command to, when empty broadcast to all workers. routing_key (str): Optional routing key. options (Dict): Additional options as supported by :meth:`kombu.entity.Queue.from_dict`. See Also: :meth:`broadcast` for supported keyword arguments. add_consumer)queueexchange exchange_type routing_keyrN)r)rEdict)r9rrrroptionsr2rGrrrr!s zControl.add_consumercKs|j d|d|id|S)zTell all (or specific) workers to stop consuming from ``queue``. See Also: Supports the same arguments as :meth:`broadcast`. cancel_consumerrrN)rr)r9rr2rGrrrrFszControl.cancel_consumercKs |j d|||d|d|S)aSTell workers to set time limits for a task by type. Arguments: task_name (str): Name of task to change time limits for. soft (float): New soft time limit (in seconds). hard (float): New hard time limit (in seconds). **kwargs (Any): arguments passed on to :meth:`broadcast`. time_limit)rhardsoftrDr2N)rr)r9rrrr2rGrrrrPs zControl.time_limitcK|j di|d|S)zTell all (or specific) workers to enable events. See Also: Supports the same arguments as :meth:`broadcast`. enable_eventsrN)rrr9r2rGrrrrdzControl.enable_eventscKr)zTell all (or specific) workers to disable events. See Also: Supports the same arguments as :meth:`broadcast`. disable_eventsrN)rrrrrrrmrzControl.disable_eventsr&cK|j dd|i|d|S)zTell all (or specific) workers to grow the pool by ``n``. See Also: Supports the same arguments as :meth:`broadcast`. pool_growrlrN)rrr9rlr2rGrrrrvszControl.pool_growcKr)zTell all (or specific) workers to shrink the pool by ``n``. See Also: Supports the same arguments as :meth:`broadcast`. pool_shrinkrlrN)rrrrrrrszControl.pool_shrinkcKs|j d||d|d|S)z}Change worker(s) autoscale setting. See Also: Supports the same arguments as :meth:`broadcast`. autoscale)maxminrN)rr)r9rrr2rGrrrrszControl.autoscalecKr)zlShutdown worker(s). See Also: Supports the same arguments as :meth:`broadcast` shutdownrN)rrrrrrrrzControl.shutdowncKs |j d|||d|d|S)aRestart the execution pools of all or specific workers. Keyword Arguments: modules (Sequence[str]): List of modules to reload. reload (bool): Flag to enable module reloading. Default is False. reloader (Any): Function to reload a module. destination (Sequence[str]): List of worker names to send this command to. See Also: Supports the same arguments as :meth:`broadcast` pool_restart)modulesreloadreloaderrN)rr)r9rrrr2rGrrrrszControl.pool_restartcKr)zTell worker(s) to send a heartbeat immediately. See Also: Supports the same arguments as :meth:`broadcast` heartbeatrN)rrrrrrrrzControl.heartbeatc  Ks|j|?} t|p ifi| }| r.| r.|| j|||||||| | | d WdS|| j|||||||| dWdS1sHwYdS)aBroadcast a control command to the celery workers. Arguments: command (str): Name of command to send. arguments (Dict): Keyword arguments for the command. destination (List): If set, a list of the hosts to send the command to, when empty broadcast to all workers. connection (kombu.Connection): Custom broker connection to use, if not set, a connection will be acquired from the pool. reply (bool): Wait for and return the reply. timeout (float): Timeout in seconds to wait for the reply. limit (int): Limit number of replies. callback (Callable): Callback called immediately for each reply received. pattern (str): Custom pattern string to match matcher (Callable): Custom matcher to run the pattern to match )channelr7r8N)r)r1rrr _broadcast)r9rFrDr2r5r$r3r6r4rr7r8 extra_kwargsrrrrrEs   $zControl.broadcastr0)NN)Nr/)NrNNN)NNN)r&N)NFNN) NNNFr/NNNNN)rmrnrorprr:r(r rr discard_allrrrrrZrrrrrrrrrrrrrErrrrrsL       %      r)"rp __future__rrrbilliard.commonr kombu.matcherr kombu.pidboxrkombu.utils.compatrkombu.utils.functionalr kombu.utils.objectsr celery.exceptionsr celery.fiver celery.utils.logr celery.utils.textr__all__rmr*rrr.objectrrrrrrs(          c