li'VddlZddlZddlZddlZddlZddlmZmZmZm Z m Z m Z ddl m Z ddlmZddlmZmZmZmZmZmZmZddlmZddlmZmZej:ej<eZ e jCejDdZ#d Z$d d hZ%d d hZ&GddZ'GddeZ(Gdde(Z)Gdde*Z+y)N)OptionalUnionTupleDictcastCallable)Queue)SelfDescribingJson) PayloadDictPayloadDictList HttpProtocolMethodSuccessCallbackFailureCallbackEmitterProtocol)one_of) EventStoreInMemoryEventStore zAiglu:com.snowplowanalytics.snowplow/payload_data/jsonschema/1-0-4httphttpsgetpostc2eZdZUeed<eed<dedefdZy) Requesterrrc8t|d|t|d|y)Nrr)setattr)selfrrs ^/var/www/html/strategist-ai/venv_dbt/lib/python3.12/site-packages/snowplow_tracker/emitters.py__init__zRequester.__init__9s fd#eS!N)__name__ __module__ __qualname__r__annotations__r r!rrr5s N M"X"H"r!rceZdZdZddddddddddiddf dededeed ed eed ee d ee d eedee e e e e ffdedeedeeefdeedeej$ddfdZe d/dededeed edef dZdeddfdZdefdZd0dZdedefdZdedefdZd0dZededefd Zd!eddfd"Zd#e ddfd$Z d#e ddfd%Z!d0d&Z"ed'eddfd(Z#dedefd)Z$d0d*Z%d0d+Z&d0d,Z'd0d-Z(d0d.Z)y)1Emitterzl Synchronously send Snowplow events to a Snowplow collector Supports both GET and POST requests rNr<endpointprotocolportmethod batch_size on_success on_failure byte_limitrequest_timeoutmax_retry_delay_secondsbuffer_capacitycustom_retry_codes event_storesessionreturnc t|tt|ttj |||||_||_| $| tt} nt| t} | |_ ||dk(rt}nd}| || kDr| }||_ ||_ |dnd|_ | |_||_||_t#j$|_t)|d|_t)|d |_| |_d|_| |_tj5d |j z|/t7t8j:t8j< |_yt7|j:|j< |_y) a :param endpoint: The collector URL. If protocol is not set in endpoint it will automatically set to "https://" - this is done automatically. :type endpoint: string :param protocol: The protocol to use - http or https. Defaults to https. :type protocol: protocol :param port: The collector port to connect to :type port: int | None :param method: The HTTP request method. Defaults to post. :type method: method :param batch_size: The maximum number of queued events before the buffer is flushed. Default is 10. :type batch_size: int | None :param on_success: Callback executed after every HTTP request in a flush has status code 200 Gets passed one argument, an array of dictionaries corresponding to the sent events' payloads :type on_success: function | None :param on_failure: Callback executed if at least one HTTP request in a flush has status code other than 200 Gets passed two arguments: 1) The number of events which were successfully sent 2) An array of dictionaries corresponding to the unsent events' payloads :type on_failure: function | None :param byte_limit: The size event list after reaching which queued events will be flushed :type byte_limit: int | None :param request_timeout: Timeout for the HTTP requests. Can be set either as single float value which applies to both "connect" AND "read" timeout, or as tuple with two float values which specify the "connect" and "read" timeouts separately :type request_timeout: float | tuple | None :param max_retry_delay_seconds: Set the maximum time between attempts to send failed events to the collector. Default 60 seconds :type max_retry_delay_seconds: int :param buffer_capacity: The maximum capacity of the event buffer. When the buffer is full new events are lost. :type buffer_capacity: int :param custom_retry_codes: Set custom retry rules for HTTP status codes received in emit responses from the Collector. By default, retry will not occur for status codes 400, 401, 403, 410 or 422. This can be overridden here. Note that 2xx codes will never retry as they are considered successful. :type custom_retry_codes: dict :param event_store: Stores the event buffer and buffer capacity. Default is an InMemoryEventStore object with buffer_capacity of 10,000 events. :type event_store: EventStore | None :param session: Persist parameters across requests by using a session object :type session: requests.Session | None N)logger)r4r:rrT)emitter repeatingFz"Emitter initialized with endpoint )rr) r PROTOCOLSMETHODSr(as_collector_urir*r-rr:r6DEFAULT_MAX_LENGTHr.r1 bytes_queuedr2r/r0 threadingRLocklock FlushTimertimer retry_timerr3 retry_delayr5inforrequestsrrrequest_method)rr*r+r,r-r.r/r0r1r2r3r4r5r6r7s rr zEmitter.__init__GsPp x#vw008T6R   &0? 0$3F '  /   &:+G(J$$$.$6DA.$$OO% = %deD'>$./"4 84==HI ?"+HLL"QD "+7;;"OD r!c>t|dkr td|jd}|jddtvr)|jd}t t |d}|d}|dk(rd}nd}| |dz|z|zS|dz|zd zt|z|zS) a :param endpoint: The raw endpoint provided by the user :type endpoint: string :param protocol: The protocol to use - http or https :type protocol: protocol :param port: The collector port to connect to :type port: int | None :param method: Either `get` or `post` HTTP method :type method: method :rtype: string r;zNo endpoint provided./z://rrz/iz#/com.snowplowanalytics.snowplow/tp2:)len ValueErrorrstripsplitr>rr str)r*r+r,r- endpoint_arrpaths rr@zEmitter.as_collector_uris$ x=1 45 5??3' >>%  #y 0#>>%0LL,q/:H#AH U?D8D <e#h.5 5e#h.4s4y@4G Gr!payloadc |j5|j'|xjtt|z c_|jdk(r7|j j |Dcic]}|t||c}n|j j ||jr|jdddycc}w#1swYyxYw)z Adds an event to the buffer. If the maximum size has been reached, flushes the buffer. :param payload: The name-value pairs for the event :type payload: dict(string:\*) Nr) rErBrPrTr-r6 add_event reached_limitflush)rrWkeys rinputz Emitter.inputsYY   ,!!SW%66!{{f$  **g+VsCWS\1B,B+VW  **73!!#    ,W  sAC)C ?AC CCc|j'|jj|jk\S|jxsd|jk\xs'|jj|jk\S)zW Checks if event-size or bytes limit are reached :rtype: bool r)r1r6sizer.rBrs rrZzEmitter.reached_limitso ?? "##((*doo= =!!&Q!O$($4$4$9$9$;t$N Or!c |j5|jjr dddy|jj }|j ||j d|_dddy#1swYyxYw)zB Sends all events in the buffer to the collector. Nr)rErH is_activer6get_events_batch send_eventsrB)rrds rr[z Emitter.flushswYY &))+ & &**;;=K   [ )  ,$%!  & & &sA9>A99Bdatacdtjd|jztjd|z |jj |j|ddi|j }|jS#tj$r}tj|Yd}~yd}~wwxYw)zZ :param data: The array of JSONs to be sent :type data: string zSending POST request to %s... Payload: %sz Content-Typezapplication/json; charset=utf-8)reheaderstimeoutN) r:rJr*debugrLrr2rKRequestExceptionwarning status_code)rreres r http_postzEmitter.http_posts  3dmmCD ]T)* ##(( ')JK,, )A}} ((  NN1  s5A==B/B**B/c^tjd|jztjd|z |jj |j||j }|jS#tj$r}tj|Yd}~yd}~wwxYw)z` :param payload: The event properties :type payload: dict(string:\*) zSending GET request to %s...rg)paramsriNrj) r:rJr*rkrLrr2rKrlrmrn)rrWrorps rhttp_getzEmitter.http_gets  2T]]BC ]W,- ##'' gt7K7K(A}} ((  NN1  s2A::B, B''B,cxtjd|jtjdy)z Calls the flush method of the base Emitter class. This is guaranteed to be blocking, not asynchronous. zStarting synchronous flush...zFinished synchronous flushN)r:rkr[rJr`s r sync_flushzEmitter.sync_flush(s'  45  01r!rnc"d|cxkxrdkScS)zz :param status_code: HTTP status code :type status_code: int :rtype: bool i,r&)rns ris_good_status_codezEmitter.is_good_status_code1sk'C''''r!evtsct|dkDrtjdt|ztj |g}g}|j dk(rRt t|j}|j|}tj|}|r||z }nQ||z }nK|j dk(r<|D]7}|j|}tj|}|r||gz }2||gz }9|jt|dkDr|j||j)t|dkDr|jt|||jr"|j|j!|y|j"j%|d|j'ytjdy)zd :param evts: Array of events to be sent :type evts: list(dict(string:\*)) rzAttempting to send %s eventsrrNFz$Skipping flush since buffer is empty)rPr:rJr(attach_sent_timestampr-r PAYLOAD_DATA_SCHEMA to_stringrqryrtr/r0 _should_retry_set_retry_delay_retry_failed_eventsr6cleanup_reset_retry_delay)rrzsuccess_eventsfailure_eventsrernrequest_succeededevts rrdzEmitter.send_events:s t9q= KK6TB C  ) )$ /NN{{f$)*=tDNNP"nnT2 $+$?$? $L!$"d*N"d*N%0C"&--"4K(/(C(CK(P%(&3%/&3%/0*s>/BQ/F/*s>/BQ/FN 3^D!!+.%%')).9  ((?'') KK> ?r!ric<|jj|y)z Set an interval at which failed events will be retried :param timeout: interval in seconds :type timeout: int | float riN)rHstartrris r_set_retry_timerzEmitter._set_retry_timergs w/r!c<|jj|y)z Set an interval at which the buffer will be flushed :param timeout: interval in seconds :type timeout: int | float rN)rGrrs rset_flush_timerzEmitter.set_flush_timerps )r!c8|jjy)z0 Abort automatic async flushing N)rGcancelr`s rcancel_flush_timerzEmitter.cancel_flush_timerxs r!eventsc:dtddfd}|D] }|| y)z Attach (by mutating in-place) current timestamp in milliseconds as `stm` param :param events: Array of events to be sent :type events: list(dict(string:\*)) :rtype: None rpr8Nc x|jdtttjdziy)Nstmi)updaterTinttime)rps rrz-Emitter.attach_sent_timestamp..updates( HHeSTYY[!1D!89: ;r!)r )rrevents rr|zEmitter.attach_sent_timestamp~s- ?  ,$%! & & &s AAA&c |jj}|j||jjFr)rrrd task_done)rrzs rrzAsyncEmitter.consume"s8::>>#D   T " JJ "r!r)r"r#r$rrTr rrrrrrrrrrrrKrr rvr[r __classcell__)rs@rrrsB"("$(0404$(GK'))-.0,0.2!GGGsm G  G SM G_-G_-GGSMG"%uUE\/B(B"CDG"%G"#G!dOGj)G (**+!G" #GR&#r!rcdeZdZdZdedefdZdedefdZd d Z defd Z deddfd Z deddfd Z y)rFzO Internal class used by the Emitter to schedule flush calls for later. r<r=c`||_||_d|_tj|_yr)r<r=rGrCrDrE)rr<r=s rr zFlushTimer.__init__.s% "04 OO% r!rir8c|j5|j dddy|j| dddy#1swYyxYw)NFrT)rErG_schedule_timerrs rrzFlushTimer.start4sK YY zz%  $$W$5    sAAA Nc|j5|j!|jjd|_dddy#1swYyxYwr)rErGrr`s rrzFlushTimer.cancel<s? YY "zz% !!#!  " " "s .AA cb|j5|jducdddS#1swYyxYwr)rErGr`s rrbzFlushTimer.is_activeBs* YY *::T) * * *s%.c|j5|jr|j|nd|_ddd|jj y#1swY$xYwr)rEr=rrGr<r[rs r_firezFlushTimer._fireFsL YY "~~$$W-!  "   " "s &AActj||j|g|_d|j_|jj y)NT)rCTimerrrGrrrs rrzFlushTimer._schedule_timerOs8__Wdjj7)D    r!r) r"r#r$rr(rr rrrrbrrr&r!rrFrF)sd&&D& Ut" *4*Utur!rF),loggingrrCrKrtypingrrrrrrrr %snowplow_tracker.self_describing_jsonr snowplow_tracker.typingr r r rrrrsnowplow_tracker.contractsrsnowplow_tracker.event_storerr basicConfig getLoggerr"r:setLevelINFOrAr}r>r?rr(robjectrFr&r!rrs$  ??D.G   8 $ GW  &/ " "oD c#7c#L))r!