o ưip@sdZddlZddlZddlZddlZddlmZmZmZm Z m Z m Z ddl m Z er1ddlmZneZddlZddlmZddlmZddlmZmZGd d d eZdS) z BETA This is the PubSub logger for GCS PubSub, this sends LiteLLM SpendLogs Payloads to GCS PubSub. Users can use this instead of sending their SpendLogs to their Postgres database. N) TYPE_CHECKINGAnyDictListOptionalUnion)StandardLoggingPayload)SpendLogsPayload)verbose_logger)CustomBatchLogger)get_async_httpx_clienthttpxSpecialProvidercseZdZ   ddeedeedeeffdd Zdeeeffdd Zd d Zd d Z de e e fdeeee ffddZZS)GcsPubSubLoggerN project_idtopic_idcredentials_pathc sddlm}|ttjd|_|ptd|_|ptd|_ |p&td|_ |jr.|j s2t dt |_tjd i|d|jit |g|_d S) a Initialize Google Cloud Pub/Sub publisher Args: project_id (str): Google Cloud project ID topic_id (str): Pub/Sub topic ID credentials_path (str, optional): Path to Google Cloud credentials JSON file r_premium_user_check)Z llm_providerZGCS_PUBSUB_PROJECT_IDZGCS_PUBSUB_TOPIC_IDZGCS_PATH_SERVICE_ACCOUNTz-Both project_id and topic_id must be provided flush_lockN)litellm.proxy.utilsrr r ZLoggingCallbackasync_httpx_clientosgetenvrrpath_service_account_json ValueErrorasyncioLockrsuper__init__ create_taskZperiodic_flush log_queue)selfrrrkwargsr __class__r^/home/app/Keep/.python/lib/python3.10/site-packages/litellm/integrations/gcs_pubsub/pub_sub.pyr s    zGcsPubSubLogger.__init__returnc sbddlm}|j|j|jddIdH\}}|jd||j|dddddd \}}d|d d }|S) z4Construct authorization headers using Vertex AI authr)vertex_chat_completionZ vertex_ai) credentialsrcustom_llm_providerNzpub-sub) model auth_headerZvertex_credentialsvertex_projectZvertex_locationZgemini_api_keystreamr*Zapi_basezBearer zapplication/json) Authorizationz Content-Type)litellmr(Z_ensure_access_token_asyncrrZ_get_token_and_url)r"r(Z _auth_headerr-r,_headersrrr&construct_request_headersEs2    z)GcsPubSubLogger.construct_request_headersc sddlm}ddlm}|z;td||dd}tjdur1|||||d}|j |n|j |t |j |j krI| IdHWdSWdStym} ztd t| d tWYd} ~ dSd} ~ ww) a Async Log success events to GCS PubSub Topic - Creates a SpendLogsPayload - Adds to batch queue - Flushes based on CustomBatchLogger settings Raises: Raises a NON Blocking verbose_logger.exception if an error occurs r)get_logging_payloadrz6PubSub: Logging - Enters logging function for model %sZstandard_logging_objectNT)r# response_obj start_timeend_timezPubSub Layer Error -  )Z1litellm.proxy.spend_tracking.spend_tracking_utilsr4rrr debuggetr0Zgcs_pub_sub_use_v1r!appendlenZ batch_sizeasync_send_batch Exception exceptionstr traceback format_exc) r"r#r5r6r7r4rZstandard_logging_payloadZspend_logs_payloaderrr&async_log_success_eventds8    z'GcsPubSubLogger.async_log_success_eventc szZz(|jsWW|jdStdt|jd|jD] }||IdHqWn!tyK}ztdt|dt WYd}~n d}~wwW|jdSW|jdS|jw)z8 Sends the batch of messages to Pub/Sub NzPubSub - about to flush z eventszPubSub Error sending batch - r8) r!clearr r9r<publish_messager>r?r@rArB)r"messagerCrrr&r=s* z GcsPubSubLogger.async_send_batchrGc sze|IdH}t|tr|}ntj|td}ddl}||dd}dd|igi}d|j d|j d }|j j |||d IdH}|j d vr[td t|jtd |jtd|j|WSty} ztd t| WYd} ~ dSd} ~ ww)z Publish message to Google Cloud Pub/Sub using REST API Args: message: Message to publish (dict or string) Returns: dict: Published message response N)defaultrzutf-8messagesdataz*https://pubsub.googleapis.com/v1/projects/z/topics/z:publish)urlr2json)zPub/Sub publish error: %szFailed to publish message: zPub/Sub response: %s)r3 isinstancer@rLdumpsbase64 b64encodeencodedecoderrrpost status_coder errortextr>r9) r"rGr2Z message_datarQZencoded_message request_bodyrKresponserCrrr&rFs2    zGcsPubSubLogger.publish_message)NNN)__name__ __module__ __qualname__rr@rrr3rDr=rr rrrF __classcell__rrr$r&rs&%. r)__doc__rrLrrAtypingrrrrrrZlitellm.types.utilsrZlitellm.proxy._typesr r0Zlitellm._loggingr Z(litellm.integrations.custom_batch_loggerr Z&litellm.llms.custom_httpx.http_handlerr r rrrrr&s