U >j>@sddlZddlZddlZddlZddlZddlZddlZddlZddlZddl m Z ddl m Z ddl mZedZGdddeZGdd d ee ZGd d d ZdS) N)contextmanager)Enum)PathZ queue_brokerc@seZdZdZdS)EmptyОчередь пустаN)__name__ __module__ __qualname____doc__r r *./modules/confluence/utils/queue_broker.pyrsrc@s$eZdZdZdZdZdZddZdS) ItemStatuspendingZ in_progressZ completedZfailedcCs|jS)N)valueselfr r r __str__szItemStatus.__str__N)rrr PENDING IN_PROGRESS COMPLETEDFAILEDrr r r r r s r c@seZdZdZd(ddZddZed)ed d d Zd d Z ddZ d*ddZ d+ddZ d,ddZ ddZddZddZd-ddZd.d d!Zd"d#Zd$d%Zd&d'ZdS)/SQLiteQueueBrokerz/dev/shmqueueNcCsz||_|dkr|j}t|}|jddd|d|jd|_t|_| t |j t d|d|jdS)NT)parentsexist_okzSQLite.z.dbuОчередь 'uK' успешно инициализирована. Путь к файлу: )name DEFAULT_PATHrmkdirjoinpath_db_path threadinglocal_local_init_dbatexitregisterclose_local_connloggerinfo)rrZdb_dirbase_dirr r r __init__$s  zSQLiteQueueBroker.__init__cCsVt|jdsNtjt|jdd}tj|_|d|d|d||j_ |jj S)uzВозвращает или создает постоянное соединение для текущего потока.conn timeoutzPRAGMA busy_timeout = 5000;zPRAGMA synchronous = NORMAL;zPRAGMA cache_size = -65536;) hasattrr"sqlite3connectstrrZRowZ row_factoryexecuter+rr+r r r _get_thread_conn4s    z"SQLiteQueueBroker._get_thread_connDEFERREDmodec csj|}z"|d||V|Wn:tk rdz |Wntjk r\YnXYnXdS)u Контекстный менеджер транзакций mode="IMMEDIATE" для записи mode="DEFERRED" для чтения zBEGIN N)r5r3commit ExceptionZrollbackr0OperationalError)rr8r+r r r _transactionFs  zSQLiteQueueBroker._transactioncCsXtjt|jdd}z4|d|d|d|d|W5|XdS)u0Инициализация базы данныхr-zPRAGMA journal_mode = WAL;a CREATE TABLE IF NOT EXISTS queue ( id TEXT PRIMARY KEY, payload TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending', result TEXT, created_at REAL NOT NULL, started_at REAL, finished_at REAL, pid INTEGER ) zBCREATE INDEX IF NOT EXISTS idx_status ON queue(status, created_at)z8CREATE INDEX IF NOT EXISTS idx_pid ON queue(pid, status)N)r0r1r2rcloser3r9r4r r r r#Ys    zSQLiteQueueBroker._init_dbc CsVtt}t}|jdd,}|dtjd|tj |dd|fW5QRX|S)u7Добавление элемента в очередь IMMEDIATEr7zd INSERT INTO queue (id, payload, status, created_at) VALUES (?, ?, 'z', ?) FZ ensure_ascii) r2uuidZuuid4timer<r3r rjsondumps)rpayloaditem_idnowr+r r r putss zSQLiteQueueBroker.putTc CsHt}t}t}d}z|jdd}|dtjd}|} | rd}| d} |dtjd tjd ||| f} | j d kr| dt | d d W5QRWSW5QRXWn<t j k r} zdt| krڂtdW5d} ~ XYnX|std|dk r,t|} | |kr,td|sttddqdS)u8Получение следующего элементаFr?r7z_ SELECT id, payload FROM queue WHERE status = 'zf' ORDER BY created_at ASC LIMIT 1 TidzT UPDATE queue SET status = 'z', started_at = ?, pid = ? WHERE id = ? AND status = 'z' rrE)rIrEzdatabase is lockeduOОчередь заблокирована, повторная попытка...Nru7Таймаут ожидания задачи истекg?g333333?)rBosgetpidr<r3r rfetchonerrowcountrCloadsr0r;r2r'debugrsleeprandomZuniform)rblockr.Z start_timeZ current_pidrGZ item_foundr+cursorrowrFupdatedeelapsedr r r getsH   "    zSQLiteQueueBroker.getc Csbt}|dk rtj|ddnd}|jdd*}|dtjdtjd|||fW5QRXdS) uPОтметка элемента как успешно обработанногоNFr@r?r7< UPDATE queue SET status = 'r', finished_at = ?, result = ? WHERE id = ? AND status = '' )rBrCrDr<r3r rr)rrFresultrG result_jsonr+r r r completeszSQLiteQueueBroker.completec Cst}t|tr(t|jt|d}n2t|tr8|}n"|dk rPdt|d}n ddd}tj|dd}|j dd *}| d t j d t j d |||fW5QRXdS) u;Отметка элемента как неудачного)errormessageN RuntimeErrorZ UnknownErrorzNo error message providedFr@r?r7rYrZr[)rB isinstancer:typerr2dictrCrDr<r3r rr)rrFr_rGZ error_datar]r+r r r fails"   zSQLiteQueueBroker.failc Cs>|jdd&}|dtjdtjd|f}W5QRX|jS)u[Восстановление зависших элементов конкретного PIDr?r7rYzu', started_at = NULL, pid = NULL WHERE pid = ? AND status = 'r[)r<r3r rrrM)rpidr+rSr r r recoverszSQLiteQueueBroker.recoverc CsD|jdd.}|dtjd}|dW5QRSQRXdS)u[Текущее количество ожидающих элементов в очередиr6r7zL SELECT COUNT(*) FROM queue WHERE status = 'r[rN)r<r3r rrLrr+rSr r r qsizes  zSQLiteQueueBroker.qsizec CsJ|jdd4}|dtjd}t|d W5QRSQRXdS)u2Проверить, пуста ли очередьr6r7zm SELECT EXISTS ( SELECT 1 FROM queue WHERE status = 'z ' ) rN)r<r3r rboolrLrhr r r emptys  zSQLiteQueueBroker.empty?c Csf|jddD}|dtjdtjd}t|d}|sLW5QRdSW5QRXt|qdS) u Блокирует выполнение, пока в очереди не останется активных элементов. Активные элементы = статус 'pending' ИЛИ 'in_progress'. r6r7z{ SELECT EXISTS ( SELECT 1 FROM queue WHERE status IN ('', 'z)') ) rTN) r<r3r rrrjrLrBrP)rZcheck_intervalr+rSZhas_active_itemsr r r waits zSQLiteQueueBroker.waitc sD|dk r dnd}|dk r|fnd}|jdd}d|d}|||}d d |D|dk rnd }|f}nd }d}|||} | } | d dk r| d nd} | d} | d} | d}d}d}| dkr| dk r|dk r|| }|dkr| |}nt| }W5QRXfdd tD}t||d<| |d<||d<||d<|S)uПолная статистика по статусам, времени и скорости обработки элементовNz WHERE pid = ?r r6r7zu SELECT status, COUNT(*) as count FROM queue z- GROUP BY status cSsi|]}|d|dqS)statuscountr .0rTr r r sz/SQLiteQueueBroker.get_stats..ao SELECT COUNT(*) as completed_count, AVG(finished_at - started_at) as avg_time, MIN(started_at) as min_started, MAX(finished_at) as max_finished FROM queue WHERE status = 'completed' AND pid = ? ac SELECT COUNT(*) as completed_count, AVG(finished_at - started_at) as avg_time, MIN(started_at) as min_started, MAX(finished_at) as max_finished FROM queue WHERE status = 'completed' Zavg_timecompleted_count min_started max_finishedrcsi|]}|j|jdqS)r)rrX)rsrpZstatsr r rtFstotalZavg_execution_time_secZtotal_execution_time_secitems_per_second)r<r3fetchallrLfloatr sumvalues)rrfZ pid_conditionZ pid_paramr+Z query_statusrSZ query_metricsZ metrics_paramZcursor_metricsZmetricsZavg_execution_timervrwrxZtotal_execution_timer{Zreportr ryr get_statss@     zSQLiteQueueBroker.get_statsc CsR|jdd<}|dtjdtjd}dd|DW5QRSQRXdS) uiВыгрузить результаты всех успешно выполненных элементовr6r7z SELECT id, status, payload, result, finished_at - started_at as duration FROM queue WHERE status IN ('rmz') c Ss\i|]T}|d|d|ddk r(|dndt|d|ddk rPt|dnddqS)rIrpZdurationNrurEr\)rpZ duration_secrEr\)rCrNrrr r r rtWs z5SQLiteQueueBroker.get_all_results..N)r<r3r rrr|rhr r r get_all_resultsOs z!SQLiteQueueBroker.get_all_resultscCsNt|jdrJ|jjdk rJz|jjWntk r<YnXt|jddS)umЗакрывает соединение текущего потока и удаляет его из кэша.r+N)r/r"r+r>r:delattrrr r r r&as z"SQLiteQueueBroker.close_local_connc Cs:t|jdr~|jjdk r~z|jjd|jjWn:tk rpz|jjWntk rjYnXYnXt|jd|j|j|jj d|j|jj dg}|D]}t dD]t}| sqz"| t d|WqWqttfk r0|dkr"t d |n td YqXqqdS) uaПолное уничтожение очереди и удаление файлов с дискаr+Nz PRAGMA wal_checkpoint(TRUNCATE);z-walz-shmu5Файл очереди успешно удален: uYНе удалось удалить файл очереди после 15 попыток: g?)r/r"r+r3r>r:rr with_namerrangeexistsunlinkr'rOPermissionErrorOSErrorr_rBrP)rZfiles_to_removeZ file_pathZattemptr r r cleanupjs4     zSQLiteQueueBroker.cleanup)rN)r6)TN)N)N)rl)N)rrr rr*r5rr2r<r#rHrXr^rergrirkrnrrr&rr r r r r!s$  6      A r)r$rCZloggingrJrQr0r rBrA contextlibrenumrpathlibrZ getLoggerr'r:rr2r rr r r r s