U .e8 @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-f4liivr4/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 d} | dk rt| nd} |D]} || dq|j t|f||| || | | d |dS) zRepublish message.Zapplication_headers content_typecontent_encodingheadersNexchange routing_key compression expiration)r/r0r1r.r,r-r2) rbody delivery_infor. propertiesr,r-popfloatpublish)producermessager/r0Z remove_propsr3infor.propsctypeencr1r2keyrrrr2s8    rcCs>|j}|dkrin|}t||||d||dddS)zMigrate single task message.Nr/r0r/r0)r4rget)r9Zbody_r:queuesr;rrrrPs   rcsfdd}|S)Ncsr|dkrdS||SNZtaskrr3r:callbacktasksrrr([sz!filter_callback..filteredr)rFrGr(rrErfilter_callbackYsrHc sVt|}t|jj|ddt|d}fdd}t|||f|d|S)z)Migrate tasks from one broker to another.F)Z auto_declarerBcsh|j}|j|j|_|j|jkr:|j|j|_|jj|jkr\|j|j|j_|dSN)channelrAnamer0r/Zdeclare)queueZ new_queuer9rBrron_declare_queueks  z'migrate_tasks..on_declare_queue)rBrO)r prepare_queuesamqpProducerrr)sourcedestZmigrateapprBkwargsrOrrNrrcs rcCst|tr|jj|S|SrJ) isinstancer"rQrB)rUqrrr _maybe_queueys  rYc 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)rY).0rM)rUrr szmove..NF)poolcs||}|rr|}t|trBt|j|jj|j}}nt|\}}t|||d| j d7_ r||rj krt dS)Nr@) rWrrZdefault_channelr/rLr0 expand_destrackr(r )r3r:retexrk) rFconnr/limit predicater9r0state transformrron_tasks$    zmove..on_task consume_from)r Zconnection_or_acquirerQrRr r) re connectionr/r0rSrUrFrdrgrVrBrhr) rUrFrcr/rdrer9r0rfrgrrs> rc Cs8z |\}}Wn"ttfk r.||}}YnX||fSrJ) TypeError ValueError)r`r/r0rarbrrrr^s  r^cCs |d|kS)z'Return true if task id equals task_id'.idr)task_idr3r:rrrrsrcCs |d|kS)z-Return true if task id is member of set ids'.rmr)idsr3r:rrrrsrcCs@t|tr|d}t|tr0tdd|D}|dkrsz!prepare_queues..)rWr"rtlistdictrIrrrrPs   rPc @sFeZdZdddZddZdd Zd d Zd d ZddZddZ dS)FiltererN?Fc s|_|_|_|_|_|_tt|p0g_t |_ | _ | _ | _ fdd| phtj D_| pxt_|_dS)Ncsg|]}tj|qSr)rYrUrur#rrr[ sz%Filterer.__init__..)rUrcfilterrdtimeout ack_messagessetr rGrPrBrFforeverrOrwrir rfaccept)r$rUrcr{rdr|r}rGrBrFrrOrirfrrVrr#r__init__s     zFilterer.__init__c Csh||Nzt|j|j|jdD]}q&Wn(tjk rDYntk rVYnXW5QRX|jS)N)r|Zignore_timeouts) prepare_consumercreate_consumerrrcr|rsocketr rf)r$_rrrstarts zFilterer.startcCs.|jjd7_|jr*|jj|jkr*tdS)Nr])rfr)rdr r$r3r:rrr update_stateszFilterer.update_statecCs |dSrJ)r_rrrr ack_message#szFilterer.ack_messagecCs|jjj|j|j|jdS)N)rBr)rUrQZ TaskConsumerrcrirr#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)rAr5rDmaprrtask_id_in_mapsz%move_by_idmap..task_id_in_maprd)rlen)rrVrrrrrts 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|dSrC)rArDrrrtask_name_in_mapsz)move_by_taskmap..task_name_in_map)r)rrVrrrrrs rcKsttjf||d|dS)N)rfr3)printMOVING_PROGRESS_FMTformat)rfr3r:rVrrr filter_statussr)rg)NNN)N)NNNNNNNN) NrzFNNNFNNNN)-rr functoolsr itertoolsrrZkomburrZ kombu.commonrZkombu.utils.encodingrZ celery.appr Zcelery.utils.nodenamesr Zcelery.utils.textr __all__r Exceptionr r rrrHrrYrr^rrrPryrrrrrrrZmove_direct_by_idmapZmove_direct_by_taskmaprrrrsj           [ Z