jsjdZddlZddlZddlZddlZddlZddlZddlZddlm Z m Z ddl m Z m Z mZmZddlmZddlmZddlmZd Zej.eZe Gd d e Ze Gd d e ZGdde Zdej:defdZGddZ ejBjDejBejFjDejFejHjDejHejJjDejJejLjDejLejNjDejNejPjDejPejRjDejRejTjDejTejVjDejVejXjDejXejZjDejZi Z.dedej:fdZ/y)u~On-disk SQLite message store used by the replay flow to persist messages that failed to reach the Opik server. DBManager exposes helpers for the full message lifecycle: register_message(s), fetch_failed_messages_batched, and replay_failed_messages (which accepts a ReplayCallback, typically Streamer.put, to re-inject messages). The manager tracks three states — undefined, initialized, closed, and error — and if the underlying database becomes unavailable, it marks itself as failed (error) and logs that resiliency features are disabled. The standalone helper db_message_to_message raises ValueError for unsupported message types. N)uniqueIntEnum)List NamedTupleOptionalIterator)messages)message_serialization)ReplayCallbackzopik_messages.dbceZdZdZdZdZdZy) MessageStatusa# Represents the status of a message. This enumeration defines the potential statuses that a message can have in a messaging system. Used for categorizing messages based on their lifecycle stage or outcome. Attributes: registered: Represents a message that has been successfully registered but not yet processed or delivered. delivered : Represents a message that has been successfully delivered to its recipient. failed: Represents a message for which delivery has failed. r N)__name__ __module__ __qualname____doc__ registered deliveredfailed/Users/manta/Documents/Projects/TheRoad-I1/backend/.venv/lib/python3.12/site-packages/opik/message_processing/replay/db_manager.pyrr s JI Frrc eZdZdZdZdZdZdZy)DBManagerStatuszRepresents various status values for a manager. Defines a set of distinct states that a manager can occupy during its lifecycle. This can be used to track and manage the status of a manager in an application. r rrN)rrrr undefined initializedclosederrorrrrrr6s IK F Errc:eZdZUdZeed<eed<eed<eed<y) DBMessagea Represents a database message entity. This class is used to encapsulate information about messages stored in or retrieved from a database. It provides a structured format for handling message data, including its unique identifier, type, JSON content, and current status. Attributes: id: The unique identifier of the message. type: The type/category of the message. json: The JSON-encoded content of the message. status: The current status of the message. idtypejsonstatusN)rrrrint__annotations__strrrrrr#r#Ds  G I I rr#messagereturncZ|j tdtj|S)NzMessage ID expected) message_id ValueErrorr serialize_message)r+s r_preprocess_registered_messager1Zs+!.// 2 27 ;;rc *eZdZdZ d"dededeedeejdee jddf d Z d#d Z d#d Zej fd ej$d eddfdZej fdeej$d eddfdZdeed eddfdZded eddfdZdedefdZdedefdZdeededefdZdedeej$fdZdedeefdZede fdZ!ede fdZ"ede fdZ#dede$eefdZ%defd Z&d eddfd!Z'y)$ DBManagerac Manages message storage, batch processing, and database operations within a replay system. This class provides functionalities to handle message registrations, batch updates, and database schema management. It ensures thread-safe operations, cleans up temporary files, and handles database connections as part of its lifecycle. N batch_sizebatch_replay_delaydb_fileconn sync_lockr,c||_||_tj|_t j |_|.tjj|jt}||_ |!tj|jd}||_|t!j"|_n||_t!j&|_|j+y)aX Initializes the Manager class, setting up the database connection, temporary directory, and other necessary properties for managing batch operations. Args: batch_size: The size of batches for processing. batch_replay_delay: The delay (in seconds) between replaying batches of messages. db_file: Path to the database file. If not provided, a temporary file will be created in a temporary directory. conn: A pre-existing database connection. If not provided, a new connection is created. NF)check_same_thread)r4r5rrr'tempfilemkdtemptmp_dirospathjoinDEFAULT_DB_FILEr6sqlite3connectr7 threadingRLock__lock__Lock _replay_mutex_create_db_schema)selfr4r5r6r7r8s r__init__zDBManager.__init__ls(%"4%// '') ?ggll4<<AG <??4<<5ID  %OO-DM%DM '^^-  rcp |j5|j5|jjdtj|_ddddddy#1swYxYw#1swYyxYw#t $r7}d|}|j|tj|dYd}~yd}~wwxYw)NaNCREATE TABLE IF NOT EXISTS messages (message_id INTEGER NOT NULL PRIMARY KEY, status INTEGER NOT NULL, message_type TEXT NOT NULL, message_json TEXT NOT NULL)z)Database schema creation failed, reason: Texc_info) rFr7executerrr' Exception_mark_as_db_failedLOGGERwarningrJexmsgs rrIzDBManager._create_db_schemas /YYII%%K#2"="=DKY /=bTBC  # #C ( NN3N . . /sK A5 A)1A A)A5A& "A))A2.A52A55 B5>-B00B5cP|jry|j5tj|_ tj d|j j|j@ tj d|jtj|jdddy#t$r"}tj d|dYd}~{d}~wwxYw#t$r-}tj d|j|dYd}~hd}~wwxYw#1swYyxYw)a Closes the current manager, releasing all associated resources. This method ensures the proper cleanup of resources such as database connections and temporary directories. It performs the operations within a thread-safe context by acquiring an internal lock. If the manager is already closed, the method will return immediately without performing any actions. NzClosing messages DB connectionz*Failed to close messages DB connection: %sTrMzCleaning temporary data dir: %rz2Failed to clean temporary data dir: %r, reason: %s) r rFrr'rRdebugr7closerPr=shutilrmtree)rJes rrYzDBManager.closes ;;  ]])00DK  => !||' LL!BDLLQMM$,,/]   @!d !LLL !% !!]sXD/B5 D-?C#5 C >CDC  D# D,#DDDDD%r+r'c|jstjdy|j5|jrtj d dddyt |}|j||j|f} |j5|jjd|ddddddy#1swYxYw#t$r}|jd|d}~wwxYw#1swYyxYw)a Registers a message into the database if the system is initialized and not closed. This method processes the given message, converts it into a JSON format, and inserts it into the database with the provided or default status. If the system is either not initialized or already closed, the message registration will be ignored. Args: message: The message object to be registered. This object must have attributes such as `message_id` and `message_type`. status: The status of the message to be registered. Defaults to `MessageStatus.registered`. z*Not initialized - register message ignoredNz)Already closed - register message ignoredINSERT INTO messages(message_id, status, message_type, message_json) VALUES (?,?,?,?) ON CONFLICT(message_id) DO UPDATE SET status = excluded.status, message_type = excluded.message_type, message_json = excluded.message_jsonzregister_messages: failed to insert messages into DB, reason: )rrRrXrFr rSr1appendr.r_r7 executemanyrPrQ)rJrcr'rar+r`rUs rregister_messageszDBManager.register_messagess LLK L  ]]{{PQ] F)=gF  **,,$ * YYII))f%]$Y  ''TUWTXY  1]sG#D=D C%CCC C D(C==DDD message_idsc 4|js!tjdt||y|j5|j r*tj dt|| dddy |j5|tjk(rW|Dcgc]}|f}}|jjd|}tjd|jt|nX|Dcgc]}||f}}|jjd|}tjd|jt||ddddddycc}wcc}w#1swYxYw#t$r}|jd|d}~wwxYw#1swYyxYw) a  Updates the status of a batch of messages in the messages database table or deletes the messages if their status is 'delivered'. The function ensures thread safety and performs operations only if the instance is initialized and not closed. Args: message_ids: A list of message IDs to be updated or deleted. status: The new status to be assigned to the messages or to determine if messages should be deleted when the status is 'delivered'. zENot initialized - messages batch update ignored, size: %d, status: %rNzDAlready closed - messages batch update ignored, size: %d, status: %r)DELETE FROM messages WHERE message_id = ?z7Deleted %d DB message records for %d delivered messages3UPDATE messages SET status = ? WHERE message_id = ?z@Updated %d DB message records for %d messages to have status: %rzJupdate_messages_batch: failed to update messages batch in the DB, reason: )rrRrXlenrFr rSr7rrrfrowcountrPrQ)rJrhr'r.db_idsc db_status_idsrUs rupdate_messages_batchzDBManager.update_messages_batch0s LLWK    ]]{{Z $ ] YY!8!88BM!N+J:-+!N II11G" UJJ ,DO)CNZVZ0;&)!II11Q) ^JJ ," )]"O)Y4 ''`ac`de  I]sa.F2 E(>E E A E- E8A EE( EE% !E(( F 1FF  FFr.c|jstjd||y|j5|jr!tj d|| dddy |j 5|tjk(r|j jd|fn|j jd||fddddddy#1swYxYw#t$r}|jd|d}~wwxYw#1swYyxYw)aE Updates the status of a message in the database or removes it if delivered. This method is responsible for updating the status of a message in the database or removing delivered messages along with their associated data. The operation is ignored if the instance is not initialized or already closed. Args: message_id: The unique identifier of the message to be updated. status: The new status to set for the message. If the status is `MessageStatus.delivered`, the message will be removed. z ? ORDER BY message_id LIMIT ?zTfetch_failed_messages_batched: failed to fetch failed messages from the DB, reason: Nrr rr) rFr7rOrrrPrQrer#r$)rJr4 last_seen_idbatchrowsrUrs rr}z'DBManager.fetch_failed_messages_batchedws %'E 99,,Z&--|ZHD CLL!"1v!$Q!$Q#0#7#7  . 9<9>.;> >@--m--^/)n/)/)b3N3s3j ?=K 0chx7K7K.L()1D: 5 5 5 :T : : 4 4 4&& $y/ "&Ps, # $ rr3rctj|j}|td|jt j ||j S)a Converts a database message object to a corresponding application message object. This function maps a database message type to its associated application message class using a predefined dictionary of supported message types. If the type is not supported, an exception is raised. The message is then deserialized from its JSON representation into the appropriate application message class. Args: db_message: The database message object containing the type and serialized JSON data to be deserialized. Returns: The deserialized application message object corresponding to the provided database message type. Raises: ValueError: If the database message contains an unsupported or unrecognized message type. zUnsupported message type: )json_str)SUPPORTED_MESSAGE_TYPESgetr%r/r deserialize_messager&)r message_classs rrrsR*,// @M5joo5FGHH 4 4  r)0rloggingr>rZrBr;rDrenumrrtypingrrrropik.message_processingr opik.message_processing.replayr typesr rA getLoggerrrRrrr#rr*r1r3CreateTraceMessager_UpdateTraceMessageCreateSpanMessageUpdateSpanMessage"AddTraceFeedbackScoresBatchMessage!AddSpanFeedbackScoresBatchMessage$AddThreadsFeedbackScoresBatchMessageCreateSpansBatchMessageCreateTraceBatchMessageGuardrailBatchMessage!CreateExperimentItemsBatchMessageCreateAttachmentMessagerrrrrrs  77,@!%   8 $G* g   ,i>i ..;;X=g=g 11>>@m@m $$1183S3S $$1183S3S ""//1O1O ..;;X=g=g $$1183S3S  iH4H4Hr