a xdS7 @sddZddlZddlmZddlmZmZddlmZm Z ddl m Z ddl m Z ddlmZdd lmZdd lmZd Zd ZGd ddeZGdddZd4ddZd5ddZddZeddfddZddZd6ddZddZdd Z d!d"Z!d#d$Z"Gd%d&d&Z#d7d)d*Z$d+d,Z%d-d.Z&d/d0Z'd1d2Z(eeed3Z)ee%ed3Z*ee&ed3Z+ee'ed3Z,dS)8z,Message migration tools (Broker <-> Broker).N)partial)cycleislice)Queue eventloop) maybe_declare) ensure_bytes)app_or_default) worker_direct) str_to_list) StopFilteringState republish migrate_task migrate_tasksmove task_id_eq task_id_in start_filtermove_task_by_id move_by_idmapmove_by_taskmap move_directmove_direct_by_idzGMoving task {state.filtered}/{state.strtotal}: {body[task]}[{body[id]}]c@seZdZdZdS)r z*Semi-predicate used to signal filter stop.N)__name__ __module__ __qualname____doc__rrN/var/www/html/Ranjet/env/lib/python3.9/site-packages/celery/contrib/migrate.pyr sr c@s0eZdZdZdZdZdZeddZddZ dS)r zMigration progress state.rcCs|js dSt|jS)N?) total_apxstrselfrrrstrtotal&szState.strtotalcCs$|jrd|jS|jd|jS)N^/)filteredcountr%r#rrr__repr__,s zState.__repr__N) rrrrr)r(r!propertyr%r*rrrrr s r c Cs|s gd}t|j}|j|j|j}}}|dur<|dn|}|durP|dn|}|j|j} } |dd} |D]} || dqr|jt|f||| || | d|dS)zRepublish message.)Zapplication_headers content_typecontent_encodingheadersNexchange routing_key compression)r/r0r1r.r,r-) rbody delivery_infor. propertiesr,r-poppublish) producermessager/r0Z remove_propsr2infor.propsctypeencr1keyrrrr2s&   rcCs>|j}|durin|}t||||d||dddS)zMigrate single task message.Nr/r0r/r0)r3rget)r7Zbody_r8queuesr9rrrrKs   rcsfdd}|S)Ncsr|dvrdS||SNZtaskrr2r8callbacktasksrrr(Vsz!filter_callback..filteredr)rDrEr(rrCrfilter_callbackTsrFc sVt|}t|jj|ddt|d}fdd}t|||f|d|S)z)Migrate tasks from one broker to another.F)Z auto_declarer@csh|j}|j|j|_|j|jkr:|j|j|_|jj|jkr\|j|j|j_|dSN)channelr?namer0r/Zdeclare)queueZ new_queuer7r@rron_declare_queuefs  z'migrate_tasks..on_declare_queue)r@rM)r prepare_queuesamqpProducerrr)sourcedestZmigrateappr@kwargsrMrrLrr^s rcCst|tr|jj|S|SrH) isinstancer"rOr@)rSqrrr _maybe_queuets  rWc stfdd|pgDp d} j|ddVjt f dd} t| fd| i| WdS1s0YdS) aG Find tasks by filtering them and move the tasks to a new queue. Arguments: predicate (Callable): Filter function used to decide the messages to move. Must accept the standard signature of ``(body, message)`` used by Kombu consumer callbacks. If the predicate wants the message to be moved it must return either: 1) a tuple of ``(exchange, routing_key)``, or 2) a :class:`~kombu.entity.Queue` instance, or 3) any other true value means the specified ``exchange`` and ``routing_key`` arguments will be used. connection (kombu.Connection): Custom connection to use. source: List[Union[str, kombu.Queue]]: Optional list of source queues to use instead of the default (queues in :setting:`task_queues`). This list can also contain :class:`~kombu.entity.Queue` instances. exchange (str, kombu.Exchange): Default destination exchange. routing_key (str): Default destination routing key. limit (int): Limit number of messages to filter. callback (Callable): Callback called after message moved, with signature ``(state, body, message)``. transform (Callable): Optional function to transform the return value (destination) of the filter function. Also supports the same keyword arguments as :func:`start_filter`. To demonstrate, the :func:`move_task_by_id` operation can be implemented like this: .. code-block:: python def is_wanted_task(body, message): if body['id'] == wanted_id: return Queue('foo', exchange=Exchange('foo'), routing_key='foo') move(is_wanted_task) or with a transform: .. code-block:: python def transform(value): if isinstance(value, str): return Queue(value, Exchange(value), value) return value move(is_wanted_task, transform=transform) Note: The predicate may also return a tuple of ``(exchange, routing_key)`` to specify the destination to where the task should be moved, or a :class:`~kombu.entity.Queue` instance. Any other true value means that the task will be moved to the default exchange/routing_key. csg|]}t|qSr)rW).0rK)rSrr zmove..NF)poolcs||}|rr|}t|trBt|j|jj|j}}nt|\}}t|||d| j d7_ r||rj krt dS)Nr>) rUrrZdefault_channelr/rJr0 expand_destrackr(r )r2r8retexrk) rDconnr/limit predicater7r0state transformrron_tasks"    zmove..on_task consume_from)r Zconnection_or_acquirerOrPr r) rd connectionr/r0rQrSrDrcrfrTr@rgr) rSrDrbr/rcrdr7r0rerfrrzs> rc Cs6z |\}}Wn ttfy,||}}Yn0||fSrH) TypeError ValueError)r_r/r0r`rarrrr]s  r]cCs |d|kS)z'Return true if task id equals task_id'.idr)task_idr2r8rrrrsrcCs |d|vS)z-Return true if task id is member of set ids'.rlr)idsr2r8rrrrsrcCs@t|tr|d}t|tr0tdd|D}|dursz!prepare_queues..)rUr"rslistdictrGrrrrNs   rNc @sFeZdZdddZddZdd Zd d Zd d ZddZddZ dS)FiltererN?Fc s|_|_|_|_|_|_tt|p0g_t |_ | _ | _ | _ fdd| phtj D_| pxt_|_dS)Ncsg|]}tj|qSr)rWrSrtr#rrrYsz%Filterer.__init__..)rSrbfilterrctimeout ack_messagessetr rErNr@rDforeverrMrvrhr reaccept)r$rSrbrzrcr{r|rEr@rDr~rMrhrerrTrr#r__init__s     zFilterer.__init__c Csx||Tzt|j|j|jdD]}q&Wn$tjyBYntyRYn0Wdn1sh0Y|jS)N)r{Zignore_timeouts) prepare_consumercreate_consumerrrbr{r~socketr re)r$_rrrstart s  $zFilterer.startcCs.|jjd7_|jr*|jj|jkr*tdS)Nr\)rer)rcr r$r2r8rrr update_stateszFilterer.update_statecCs |dSrH)r^rrrr ack_messageszFilterer.ack_messagecCs|jjj|j|j|jdS)N)r@r)rSrOZ TaskConsumerrbrhrr#rrrr!s zFilterer.create_consumercCs|j}|j}|j}|jr>> move_by_idmap({ ... '5bee6e82-f4ac-468e-bd3d-13e8600250bc': Queue('name'), ... 'ada8652d-aef3-466b-abd2-becdaf1b82b3': Queue('name'), ... '3a2b140d-7db1-41ba-ac90-c36a0ef4ab1f': Queue('name')}, ... queues=['hipri']) cs|jdS)NZcorrelation_id)r?r4rBmaprrtask_id_in_map{sz%move_by_idmap..task_id_in_maprc)rlen)rrTrrrrros rc sfdd}t|fi|S)aMove tasks by matching from a ``task_name: queue`` mapping. ``queue`` is the queue to move the task to. Example: >>> move_by_taskmap({ ... 'tasks.add': Queue('name'), ... 'tasks.mul': Queue('name'), ... }) cs|dSrA)r?rBrrrtask_name_in_mapsz)move_by_taskmap..task_name_in_map)r)rrTrrrrrs rcKsttjf||d|dS)N)rer2)printMOVING_PROGRESS_FMTformat)rer2r8rTrrr filter_statussr)rf)NNN)N)NNNNNNNN) NryFNNNFNNNN)-rr functoolsr itertoolsrrZkomburrZ kombu.commonrZkombu.utils.encodingrZ celery.appr Zcelery.utils.nodenamesr Zcelery.utils.textr __all__r Exceptionr r rrrFrrWrr]rrrNrxrrrrrrrZmove_direct_by_idmapZmove_direct_by_taskmaprrrrsV           [ Z