Qz>j>ddlZddlZddlZddlZddlZddlZddlZddlZddlZddl m Z ddl m Z ddl mZejdZGddeZGdd ee ZGd d ZdS) N)contextmanager)Enum)Path queue_brokerceZdZdZdS)EmptyОчередь пустаN)__name__ __module__ __qualname____doc__*./modules/confluence/utils/queue_broker.pyrrs##Drrc$eZdZdZdZdZdZdZdS) ItemStatuspending in_progress completedfailedc|jSN)valueselfs r__str__zItemStatus.__str__s zrN)r r r PENDING IN_PROGRESS COMPLETEDFAILEDrrrrrrs7GKI FrrceZdZdZddZdZeddefdZd Z d Z dd Z dd Z ddZ dZdZdZddZddZdZdZdZdS)SQLiteQueueBrokerz/dev/shmqueueNc||_||j}t|}|dd|d|jd|_t j|_| tj |j td|d|jdS)NT)parentsexist_okzSQLite.z.dbuОчередь 'uK' успешно инициализирована. Путь к файлу: )name DEFAULT_PATHrmkdirjoinpath_db_path threadinglocal_local_init_dbatexitregisterclose_local_connloggerinfo)rr'db_dirbase_dirs r__init__zSQLiteQueueBroker.__init__$s >&F<<td333 ))*BDI*B*B*BCC o''  -... HtHHx|yFHH I I I I IrcLt|jdstjt |jd}tj|_|d|d|d||j_ |jj S)uzВозвращает или создает постоянное соединение для текущего потока.conn timeoutzPRAGMA busy_timeout = 5000;zPRAGMA synchronous = NORMAL;zPRAGMA cache_size = -65536;) hasattrr.sqlite3connectstrr+Row row_factoryexecuter9rr9s r_get_thread_connz"SQLiteQueueBroker._get_thread_conn4st{F++ $?3t}#5#5rBBBD&{D  LL6 7 7 7 LL7 8 8 8 LL6 7 7 7#DK {rDEFERREDmodec#K|} |d||V|dS#t$r- |n#t j$rYnwxYwwxYw)u Контекстный менеджер транзакций mode="IMMEDIATE" для записи mode="DEFERRED" для чтения zBEGIN N)rErCcommit Exceptionrollbackr>OperationalError)rrGr9s r _transactionzSQLiteQueueBroker._transactionFs$$&&  LL$ ) ) )JJJ KKMMMMM     +       s/0A BA*)B*A<9B;A<<Bctjt|jd} |d|d|d|d||dS#|wxYw)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)r>r?r@r+rCrIcloserDs rr/zSQLiteQueueBroker._init_dbYss4=112>>>  LL5 6 6 6 LL     LL] ^ ^ ^ LLS T T T KKMMM JJLLLLLDJJLLLLs A(B((B>c >ttj}tj}|d5}|dt jd|tj |d|fdddn #1swxYwY|S)u7Добавление элемента в очередь IMMEDIATErGzd INSERT INTO queue (id, payload, status, created_at) VALUES (?, ?, 'z', ?) F ensure_asciiN) r@uuiduuid4timerMrCrrjsondumps)rpayloaditem_idnowr9s rputzSQLiteQueueBroker.putssdjll##ikk   K  0 0 JD LL * 24:gEBBBCH J J J J J J J J J J J J J J J J J J s rLr@r3debugrsleeprandomuniform)rblockr< start_time current_pidr] item_foundr9cursorrowr\updatedeelapseds rgetzSQLiteQueueBroker.getsjY[[ ikk / 7)++CJ p&&K&88D!\\+)3);+++F !//++C%) "%d)"&,,0+5+A00 9C8J 000 ";8 #:#:#+a//&)$i+/:c)n+E+E$$-4+ p p p's1vv55 noooooooo p  97888")++ 2g%% YZZZ 7 6>$55666_/ 7sBDB%D9 D DDDDDE#-,EE#c&tj}|tj|dnd}|d5}|dt jdt jd|||fddddS#1swxYwYdS) uPОтметка элемента как успешно обработанногоNFrTrRrS< UPDATE queue SET status = 'r', finished_at = ?, result = ? WHERE id = ? AND status = '' )rXrYrZrMrCrrr)rr\resultr] result_jsonr9s rcompletezSQLiteQueueBroker.completesikk@F@Rdje<<<YZZJj%@@@   K  0 0 .D LL)0 -7,B  {G,  . . . . . . . . . . . . . . . . . . . . .s,5C..C25C2c|d5}|dtjdtjd|f}dddn #1swxYwY|jS)u[Восстановление зависших элементов конкретного PIDrRrSruzu', started_at = NULL, pid = NULL WHERE pid = ? AND status = 'rwN)rMrCrrrrd)rpidr9rns rrecoverzSQLiteQueueBroker.recovers   K  0 0 D\\#)1## .8-C ###  F               s3AAAc|d5}|dtjd}|dcdddS#1swxYwYdS)u[Текущее количество ожидающих элементов в очередиrFrSzL SELECT COUNT(*) FROM queue WHERE status = 'rwrN)rMrCrrrcrr9rns rqsizezSQLiteQueueBroker.qsizes   J  / / (4\\#!+!3###F??$$Q'  ( ( ( ( ( ( ( ( ( ( ( ( ( ( ( ( ( (s=A!!A%(A%c|d5}|dtjd}t |d cdddS#1swxYwYdS)u2Проверить, пуста ли очередьrFrSzm SELECT EXISTS ( SELECT 1 FROM queue WHERE status = 'z ' ) rN)rMrCrrboolrcrs remptyzSQLiteQueueBroker.emptys   J  / / 24\\#&0%7###F FOO--a0111 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2sA A//A36A3?cZ |d5}|dtjdtjd}t |d}|s ddddS dddn #1swxYwYtj|) u Блокирует выполнение, пока в очереди не останется активных элементов. Активные элементы = статус 'pending' ИЛИ 'in_progress'. TrFrSz{ SELECT EXISTS ( SELECT 1 FROM queue WHERE status IN ('', 'z)') ) rN) rMrCrrrrrcrXrg)rcheck_intervalr9rnhas_active_itemss rwaitzSQLiteQueueBroker.waits! '"" "33 t',6+=''DNCY''' $((9(9!(<#=#= '     J~ & & & 'sAB  BBc|dnd}||fnd}|d5}d|d}|||}d |D|d }|f}nd }d}|||} | } | d | d nd } | d} | d} | d}d }d }| dkr$| "| || z }|dkr| |z }nt | }dddn #1swxYwYfdt D}t ||d<| |d<||d<||d<|S)uПолная статистика по статусам, времени и скорости обработки элементовNz WHERE pid = ?rrFrSzu SELECT status, COUNT(*) as count FROM queue z- GROUP BY status c,i|]}|d|dS)statuscountr.0ros r z/SQLiteQueueBroker.get_stats..s"NNNSS]CLNNNrao 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' avg_timecompleted_count min_started max_finishedrcRi|]#}|j|jd$S)r)rrs)rrstatss rrz/SQLiteQueueBroker.get_stats..Fs-TTTv&, &, : :TTTrtotalavg_execution_time_sectotal_execution_time_secitems_per_second)rMrCfetchallrcfloatrsumvalues)rr pid_condition pid_paramr9 query_statusrn query_metrics metrics_paramcursor_metricsmetricsavg_execution_timerrrtotal_execution_timerreportrs @r get_statszSQLiteQueueBroker.get_statss +.? !oSFF2   J  / /1 >4* L\\, ::FNNFOO2L#& " ""{'>1 >1 >1 >1 >1 >1 >1 >1 >1 >1 >1 >1 >1 >1 >fUTTTTTTfmmoo..w+='(-A)*%5!" sC C::C>C>c|d5}|dtjdtjd}d|DcdddS#1swxYwYdS)uiВыгрузить результаты всех успешно выполненных элементовrFrSz SELECT id, status, payload, result, finished_at - started_at as duration FROM queue WHERE status IN ('rz') c i|]`}|d|d|d|dndtj|d|dtj|dnddaS)r`rdurationNrr[rx)r duration_secr[rx)rYrers rrz5SQLiteQueueBroker.get_all_results..Ws D !(m7::7RC OOX[#z#i.99;>x=;TdjX777Z^ rN)rMrCrrr rrs rget_all_resultsz!SQLiteQueueBroker.get_all_resultsOs   J  / / 4\\#$.#7##>H=N###F "??,,                   sAA22A69A6ct|jdrS|jjI |jjn#t$rYnwxYwt |jddSdSdS)umЗакрывает соединение текущего потока и удаляет его из кэша.r9N)r=r.r9rPrJdelattrrs rr2z"SQLiteQueueBroker.close_local_connas 4; ' ' )DK,<,H   &&((((     DK ( ( ( ( (  ) ),H,HsA AAcvt|jdr|jj |jjd|jjn@#t $r3 |jjn#t $rYnwxYwYnwxYwt |jd|j|j|jj d|j|jj dg}|D]}tdD]}| sn | td|nO#tt f$r;|dkrtd |nt%jd YwxYwdS) uaПолное уничтожение очереди и удаление файлов с дискаr9Nz PRAGMA wal_checkpoint(TRUNCATE);z-walz-shmu5Файл очереди успешно удален: uYНе удалось удалить файл очереди после 15 попыток: g?)r=r.r9rCrPrJrr+ with_namer'rangeexistsunlinkr3rfPermissionErrorOSErrorr|rXrg)rfiles_to_remove file_pathattempts rcleanupzSQLiteQueueBroker.cleanupjs  4; ' ' )DK,<,H   (()KLLL  &&((((   K$**,,,, D  DK ( ( ( M M # #t}'9$?$?$? @ @ M # #t}'9$?$?$? @ @  ) ( (I 99 ( ( ''))E($$&&&LL!dYb!d!deeeE'1((("}} &MBK&M&MNNNN 3 ( ( (sH=A!! B,B  B BBBBB61E))A F54F5)r#N)rF)TNr)r)r r r r(r7rErr@rMr/r^rsrzrrrrrrrr2rrrrr"r"!sJLIIII    $^$4   47474747l . . . ......   ((( 2 2 2''''&????B$)))!(!(!(!(!(rr")r0rYloggingrarhr>r,rXrV contextlibrenumrpathlibr getLoggerr3rJrr@rr"rrrrs3   %%%%%%  > * *     I   dj(j(j(j(j(j(j(j(j(j(r