U Z+dS7 @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__rr:/tmp/pip-unpacked-wheel-ucduq0nd/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|sddddg}t|j}|j|j|j}}}|dkr@|dn|}|dkrT|dn|}|j|j} } |dd} |D]} || dqv|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.propsctypeencr1keyrrrr2s2   rcCs>|j}|dkrin|}t||||d||dddS)zMigrate single task message.Nr/r0r/r0)r3rget)r7Zbody_r8queuesr9rrrrKs   rcsfdd}|S)Ncsr|dkrdS||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|ddTjt f dd} t| fd| i| W5QRSQRXdS) 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 szmove..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) rc connectionr/r0rQrSrDrbrerTr@rfr) rSrDrar/rbrcr7r0rdrerrzs> rc Cs8z |\}}Wn"ttfk r.||}}YnX||fSrH) TypeError ValueError)r^r/r0r_r`rrrr\s  r\cCs |d|kS)z'Return true if task id equals task_id'.idr)task_idr2r8rrrrsrcCs |d|kS)z-Return true if task id is member of set ids'.rkr)idsr2r8rrrrsrcCs@t|tr|d}t|tr0tdd|D}|dkrsz!prepare_queues..)rUr"rrlistdictrGrrrrNs   rNc @sFeZdZdddZddZdd Zd d Zd d ZddZddZ dS)FiltererN?Fc s|_|_|_|_|_|_tt|p0g_t |_ | _ | _ | _ fdd| phtj D_| pxt_|_dS)Ncsg|]}tj|qSr)rWrSrsr#rrrYsz%Filterer.__init__..)rSrafilterrbtimeout ack_messagessetr rErNr@rDforeverrMrurgr rdaccept)r$rSraryrbrzr{rEr@rDr}rMrgrdr~rTrr#r__init__s     zFilterer.__init__c Csh||Nzt|j|j|jdD]}q&Wn(tjk rDYntk rVYnXW5QRX|jS)N)rzZignore_timeouts) prepare_consumercreate_consumerrrarzr}socketr rd)r$_rrrstart s zFilterer.startcCs.|jjd7_|jr*|jj|jkr*tdS)Nr[)rdr)rbr r$r2r8rrr update_stateszFilterer.update_statecCs |dSrH)r]rrrr ack_messageszFilterer.ack_messagecCs|jjj|j|j|jdS)N)r@r~)rSrOZ TaskConsumerrargr~r#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_maprb)rlen)rrTrrrrros rc sfdd}t|f|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)rdr2)printMOVING_PROGRESS_FMTformat)rdr2r8rTrrr filter_statussr)re)NNN)N)NNNNNNNN) NrxFNNNFNNNN)-rr functoolsr itertoolsrrZkomburrZ kombu.commonrZkombu.utils.encodingrZ celery.appr Zcelery.utils.nodenamesr Zcelery.utils.textr __all__r Exceptionr r rrrFrrWrr\rrrNrwrrrrrrrZmove_direct_by_idmapZmove_direct_by_taskmaprrrrsj           [ Z