jKjEdZddlZddlZddlZddlZddlmZmZmZddl m Z ddl m Z dZ dZd Zej d ZGd d Zd gZy) aL Integration for `celery`_ to capture task lifecycle events and exceptions with PostHog. .. _celery: https://pypi.org/project/celery/ Features: - Hooks into Celery signals to automatically capture task lifecycle events (started, success, failure, retry, published) and exceptions. - Lifecycle events include Celery-specific properties such as task ID, task name, queue, retry count, duration, Celery version etc. - Any custom events captured inside a task (via ``client.capture``) are automatically enriched with the same Celery-specific properties via context tags. - Propagates PostHog context (distinct ID, session ID, tags) from the producer process to the worker process. Supports Celery 4.0+ (Message Protocol Version 2). Usage ----- .. code-block:: python from posthog import Posthog from posthog.integrations.celery import PosthogCeleryIntegration # ... init Posthog client integration = PosthogCeleryIntegration() integration.instrument() # ... publish tasks or run workers ... integration.shutdown() posthog.shutdown() See ``examples/celery_integration.py`` for a complete working example. Supported task states for event emission: - ``published`` - ``started`` - ``success`` - ``failure`` - ``retry`` Event properties: All lifecycle and exception events include the following properties: - ``celery_task_id`` -- unique task ID - ``celery_task_name`` -- registered task name - ``celery_state`` -- lifecycle state (started, success, failure, etc.) - ``celery_hostname`` -- worker hostname - ``celery_exchange`` -- broker exchange - ``celery_routing_key`` -- broker routing key - ``celery_queue`` -- broker queue name - ``celery_retry_count`` -- number of retries so far - ``celery_version`` -- installed Celery library version - ``celery_task_duration_ms`` -- task wall-clock duration in milliseconds (present on terminal states: success, failure, retry) Additional properties on specific states: - **failure**: ``error_type``, ``error_message`` - **retry**: ``celery_reason`` N)AnyCallableOptional)contexts)ClientzX-POSTHOG-DISTINCT-IDzX-POSTHOG-SESSION-IDzX-POSTHOG-CONTEXT-TAGSposthogceZdZdZ d&deededededeeeee ee fgeff dZ d'd Z d'd Z d'd Zd'd Zd'dZdZdZdZdZdZdZ d(dedee ee fd dfdZde d dfdZde d e ee ffdZde d e ee ffdZ d)d e ee ffdZde de ee fd dfdZdeede ee fd efd Zd!ed"e ee fd dfd#Zd$ed dfd%Z y)*PosthogCeleryIntegrationawCelery integration that captures task lifecycle events and exceptions. Args: client: Optional ``Client`` instance. When provided, all events and exceptions are captured through this client rather than the global ``posthog`` module. Don't skip this if using a custom flag definition cache provider, and pass the custom ``Client`` instance here initialized with the custom provider so fork safety for that provider is handled correctly. capture_exceptions: Whether to capture task exceptions via ``capture_exception`` (default ``True``). capture_task_lifecycle_events: Whether to emit lifecycle events of the task such as "started", "success", "failure" etc. (default ``True``). propagate_context: Whether to propagate PostHog context (distinct ID, session ID, tags) from the producer to the worker via task headers (default ``True``). task_filter: Optional callback ``(task_name, task_properties) -> bool`` expected to return ``False`` if a given task should not be tracked. Nclientcapture_exceptionscapture_task_lifecycle_eventspropagate_context task_filterc||_||_||_||_||_d|_d|_d|_d|_yNF) r r rrr _instrumented _shut_down_signals_celery_version)selfr r rrrs l/Users/ahmed/devFolder/Ultron/claude-voice/.venv/lib/python3.12/site-packages/posthog/integrations/celery.py__init__z!PosthogCeleryIntegration.__init__hsI "4-J*!2&"'+ .2returnc|jryddlm}ddlm}d|_||_||_|jj|jd|jj|jd|jj|jd|jj|jd|j j|j"d|j$j|j&d|j(j|j*dt-j.|j0d|_y)a3Connect Celery signal handlers to capture task events and exceptions. Call this after initializing the PostHog client and this integration. If Celery runs on a single host, reinstrumenting in worker children is not strictly necessary because the PostHog client and this integration are fork-safe. If Celery workers run across multiple hosts, each worker process must initialize PostHog, this integration, and call ``instrument()``. Celery provides ``worker_process_init`` signal to help with this. Nr)signals) __version__F)weakT)rceleryrrrrr task_prerunconnect_on_task_prerun task_success_on_task_success task_failure_on_task_failure task_retry_on_task_retrybefore_task_publish_on_before_task_publishafter_task_publish_on_after_task_publishworker_process_shutdown_on_worker_process_shutdownatexitregistershutdown)rrcelery_versions r instrumentz#PosthogCeleryIntegration.instrument{s$    "8 -##D$8$8u#E$$T%:%:$G$$T%:%:$G""4#6#6U"C##++D,H,Hu+U""**4+F+FU*S''//  , ,5 0   &!rc|jr |jsy|jjj|j|jj j|j |jjj|j|jjj|j|jjj|j|jjj|j|jjj|j d|_d|_yr)rrr! disconnectr#r$r%r&r'r(r)r*r+r,r-r.r/rs r_disconnect_signalsz,PosthogCeleryIntegration._disconnect_signalss!!  !!,,T-A-AB ""--d.C.CD ""--d.C.CD   ++D,?,?@ ))44T5Q5QR ((33D4O4OP --88  , ,  "rcb|jtj|jy)zDisconnect Celery signal handlers and unregister exit cleanup. Do not use directly, call `shutdown()` instead. N)r8r0 unregisterr2r7s r uninstrumentz%PosthogCeleryIntegration.uninstruments"   "$--(rc|jry |j|jr|jjn ddlm}||j d|_y#t $rtjdYywxYw)aDisconnect all signal handlers registered by ``instrument()``, flush all pending events and cleanly shutdown the integration. ``shutdown()`` is also registered on ``worker_process_shutdown`` and ``atexit`` signals, but there is no guarantee those will always be called, so we strongly recommend calling it manually when the integration is no longer needed to avoid data loss. Nr)flushTz.Failed to shut down PostHog Celery integration) rr8r r=r; Exceptionlogger exception)rr=s rr2z!PosthogCeleryIntegration.shutdownsr ??  O  $ $ &{{ !!#$    "DO O   M N OsAA++B  B c$|jyN)r2rargskwargss rr/z4PosthogCeleryIntegration._on_worker_process_shutdowns  rc |jsy|jd}t|tsyt j }t j }t j}i}|r ||t<|r ||t<|r"tj|t|t<|r3|j||jdij|yy#t $rt"j%dYywxYw)Nheaders)defaultz:Failed to propagate PostHog context in before_task_publish)rget isinstancedictrget_context_distinct_idget_context_session_idget_tagsCONTEXT_DISTINCT_ID_HEADERCONTEXT_SESSION_ID_HEADERjsondumpsstrCONTEXT_TAGS_HEADERupdate setdefaultr?r@rA)rrErFrH distinct_id session_idtagsposthog_headerss rr+z0PosthogCeleryIntegration._on_before_task_publishs ))jj+Ggt,"::I :;=G 9:7;zz$PS7T 34/ ""9b188I    L  s C!CB,CD?Dc  |jsy|jd}|jd}t|tr|jdnd}||d|jd|jddt|tr|jdnd|jd}|j ||r|j d | yy#t$rtjd YywxYw) NsenderrHid publishedexchange routing_keyretries)celery_task_idcelery_task_name celery_statecelery_exchangecelery_routing_keycelery_hostnamecelery_retry_countr3zcelery task published propertiesz;Failed to capture Celery after_task_publish lifecycle event) rrJrKrLr _should_track_capture_eventr?r@rA)rrErFr]rHtask_idsender_propertiess rr-z/PosthogCeleryIntegration._on_after_task_publishs 55ZZFjj+G+5gt+Dgkk$'$G#*$* +#)::j#9&,jj&?#'gt,'.kk)&<"&"6"6 ! !!&*;<##+8I$=    M  s CB>CC0/C0c d} |jd}|sy|jd}t|dd}|j|}|j||d}|jd} |Rt j dd|j }|j||_tj|_ |j|i||} | jD]\} } t j| | |jr'|j!| |r|j#d | yyy#t$$r@t&j)d |& |j+dddYy#t$$rYYywxYwYywxYw) Nrnr]requeststartedr]rnstaterdTF)freshr r zcelery task startedrjz$Failed to process Celery task_prerun)rJgetattr_extract_propagated_tags_build_task_propertiesr new_contextr __enter__ _posthog_ctxtime monotonic_posthog_start_apply_propagated_identityitemstagrrlrmr?r@rA__exit__) rrErFcontext_managerrnr]rq context_tagstask_properties task_name merged_tagskeyvalues rr#z(PosthogCeleryIntegration._on_task_preruns) jj+GZZ)Ffi6G88AL"99:O (++,>?I""*"6"6',;;#  ))+'6$)-)9&  + +G 4=_= =K)//1 U S%(211d6H6H?7##$9o#V71    C D*#,,T4> + s5D9DD99 FE// E<8F;E<<FFc(|jdi|y)N)success_handle_task_endrDs rr%z)PosthogCeleryIntegration._on_task_successE262rc(|jdi|y)N)failurerrDs rr'z)PosthogCeleryIntegration._on_task_failureHrrc b|j dddt|jdii|y)Nextra_properties celery_reasonreason)retry)rrTrJrDs rr)z'PosthogCeleryIntegration._on_task_retryKs>   VZZ%9!:   rrtrc |jd}t|dd} |jd}| t|dd}|j|||}|r|j||j |||jd}|rCt |j |d<t||d<|jr|j||jd } |jr(|j| |r|jd || t|d d} | | j!dddyy#t$rtjd |YEwxYw#t|d d} | | j!dddwwxYw)Nr]rqrnr^rsrA error_type error_messagerdz celery task rjz!Failed to process Celery %s stater{)rJrvrxrV _add_durationtype__name__rTr _capture_exceptionrrlrmr?r@rAr) rrtrrFr]rqrnrrArctxs rrz)PosthogCeleryIntegration._handle_task_endTs H%&)T2 /jj+G!'46"99:O  &&'78   w 8 ;/I04Y0H0H -36y>0**++I6'++,>?I11d6H6H?7##l5'$:#W'>48C T4.  I   @% H I'>48C T4.s$C4D88EEEE%Frqc|j|}|jt}|rtjt ||jt }|rtjt |yyrC)_extract_headersrJrPridentify_contextrTrQset_context_session)rrqrHrXrYs rrz3PosthogCeleryIntegration._apply_propagated_identitys`''0kk"<=   % %c+&6 7[[!:;   ( (Z 9 rc|j|}|jt}t|tt t fsiS tj|}t|tr|SiS#t$ricYSwxYwrC) rrJrUrKrTbytes bytearrayrRloadsr?rL)rrqrHraw_tagsparseds rrwz1PosthogCeleryIntegration._extract_propagated_tagssv''0;;23(S%$;<I ZZ)F fd #M  I sA.. A<;A<c|iSt|dd}t|tr|St|tr#|jd}t|tr|SiS)NrH)rvrKrLrJ)rrqrH dict_headerss rrz)PosthogCeleryIntegration._extract_headerssX ?I '9d3 gt $N gt $";;y1L,-## rc *t|dd}t|dd}t|tr|ni}|t|dd|t|dd|jd|jd|jdt|dd|jd }|S) Nrq delivery_infonamehostnamer`raqueuerb) rcrdrerhrfrg celery_queuerir3)rvrKrLrJr)rr]rnrtrqrrks rrxz/PosthogCeleryIntegration._build_task_propertiess &)T2$? )3M4)H b & ' =!&w DA,00<"/"3"3M"B)--g6")'9d"C"22  rrcrt|dd}|(ttj|z dzd|d<yy)Nr~g@@celery_task_duration_ms)rvroundr|r})rrqr start_times rrz&PosthogCeleryIntegration._add_durationsBW&6=  !9>!J.&8!:O5 6 "rrcR|jrt|j||Sy)NT)rbool)rrrs rrlz&PosthogCeleryIntegration._should_tracks(   ((ODE Ereventrkcx|jr|jj||yddlm}|||y)Nrjr)capture)r rr )rrrkrs rrmz'PosthogCeleryIntegration._capture_events- ;; KK  *  = ' Ej 1rrAcp|jr|jj|yddlm}||y)Nr)capture_exception)r rr )rrArs rrz+PosthogCeleryIntegration._capture_exceptions& ;; KK ) )) 4 1 i (r)NTTTN)rNrC)NNN)!r __module__ __qualname____doc__rrrrrTrLrrr4r8r;r2r/r+r-r#r%r'r)rrrwrrxrrlrmr?rrrr r Ss,$(#'.2"&QU 3 3!3(, 3  3 h tCH~'F'LMN 3&!"F#$)O4 DB+Z33 6:)/)/#4S>2)/  )/V:#:$:S#X S#X(  c3h .S4S>d!#9=c3h 2C2T#s(^22)I)$)rr )rr0rRloggingr|typingrrrr>rr rrPrQrU getLoggerr@r __all__rrrrsd?B  **52.   9 %K)K)^  r