Bj>ddlZddlZddlZddlZddlZddlZddlZddlZddlZddl m Z ddl m Z ddl mZejdZGddeZGdd ee ZGd d Zy) N)contextmanager)Enum)Path queue_brokerceZdZdZy)EmptyОчередь пустаN)__name__ __module__ __qualname____doc__*./modules/confluence/utils/queue_broker.pyrrs#rrc"eZdZdZdZdZdZdZy) ItemStatuspending in_progress completedfailedc|jSN)valueselfs r__str__zItemStatus.__str__s zzrN)r r r PENDING IN_PROGRESS COMPLETEDFAILEDrrrrrrsGKI FrrceZdZdZddZdZeddefdZdZ dZ dd Z dd Z dd Z d Zd ZdZddZddZdZdZdZy)SQLiteQueueBrokerz/dev/shmNc||_| |j}t|}|jdd|j d|jd|_t j|_|jtj|jtjd|d|j y)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<td3 ))GDII;c*BC oo'  --. &tf,wx|yFyFxGH Irc`t|jdstjt |j d}tj |_|jd|jd|jd||j_ |jjS)uzВозвращает или создает постоянное соединение для текущего потока.conn timeoutzPRAGMA busy_timeout = 5000;zPRAGMA synchronous = NORMAL;zPRAGMA cache_size = -65536;) hasattrr-sqlite3connectstrr*Row row_factoryexecuter8rr8s r_get_thread_connz"SQLiteQueueBroker._get_thread_conn4s|t{{F+??3t}}#5rBD&{{D  LL6 7 LL7 8 LL6 7#DKK {{rmodec#K|j} |jd|||jy#t$r, |j #t j $rYwxYwwxYww)u Контекстный менеджер транзакций mode="IMMEDIATE" для записи mode="DEFERRED" для чтения zBEGIN N)rDrBcommit Exceptionrollbackr=OperationalError)rrEr8s r _transactionzSQLiteQueueBroker._transactionFsy$$&  LL6$ )J KKM    ++    s=A5(=A5 A2AA2A.+A2-A..A22A5cLtjt|jd} |j d|j d|j d|j d|j |j y#|j 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*rBrGcloserCs rr.zSQLiteQueueBroker._init_dbYsvs4==12>  LL5 6 LL   LL] ^ LLS T KKM JJLDJJLs ABB#c 0ttj}tj}|j d5}|j dt jd|tj|d|fddd|S#1swY|SxYw)u7Добавление элемента в очередь IMMEDIATErEzd INSERT INTO queue (id, payload, status, created_at) VALUES (?, ?, 'z', ?) F ensure_asciiN) r?uuiduuid4timerKrBrrjsondumps)rpayloaditem_idnowr8s rputzSQLiteQueueBroker.putssdjjl#iik   K  0 JD LL * 2 234 4::gEBCH J J   J s =B  Bcptj}tj} tj}d} |jd5}|j dt j d}|j} | rud}| d} |j dt jd t j d ||| f} | jd kDr'| dtj| d d cdddSddd|s t#d|'tj|z } | |k\r t#d|s)tj$t'j(dd?#1swYlxYw#tj$r-} dt| vrtj!dYd} ~ d} ~ wwxYw)u8Получение следующего элементаTFrPrQz_ SELECT id, payload FROM queue WHERE status = 'zf' ORDER BY created_at ASC LIMIT 1 idzT UPDATE queue SET status = 'z', started_at = ?, pid = ? WHERE id = ? AND status = 'z' rrY)r^rYNzdatabase is lockeduOОчередь заблокирована, повторная попытка...r u7Таймаут ожидания задачи истекg?g333333?)rVosgetpidrKrBrrfetchonerrowcountrWloadsr=rJr?r2debugrsleeprandomuniform)rblockr; start_time current_pidr[ item_foundr8cursorrowrZupdatedeelapseds rgetzSQLiteQueueBroker.getsYY[ iik ))+CJ p&&K&8D!\\/))3););(<=+F !//+C%) "%d)"&,,4++5+A+A*BC89C8J8J7KL 0 ";8 #:#++a/&)$i+/::c)n+E$-@788"))+ 2g% YZZ 6>>$56_ 4++ p's1v5 noo ps7E5B!E)4 E5>E5)E2.E55F5#F00F5c tj}|tj|dnd}|jd5}|j dt j dt jd|||fdddy#1swYyxYw) uPОтметка элемента как успешно обработанногоNFrRrPrQ< UPDATE queue SET status = 'r', finished_at = ?, result = ? WHERE id = ? AND status = '' )rVrWrXrKrBrrr)rrZresultr[ result_jsonr8s rcompletezSQLiteQueueBroker.completesiik@F@RdjjeYZJjj%@   K  0 .D LL)0012,-7,B,B+CD  {G,  . . . .s 9CC$c|jd5}|jdtjdtjd|f}ddd|j S#1swYj SxYw)u[Восстановление зависших элементов конкретного PIDrPrQrszu', started_at = NULL, pid = NULL WHERE pid = ? AND status = 'ruN)rKrBrrrrb)rpidr8rls rrecoverzSQLiteQueueBroker.recovers   K  0 D\\')1123-.8-C-C,DE #  F  s 7AA2c|jd5}|jdtjd}|j dcdddS#1swYyxYw)u[Текущее количество ожидающих элементов в очередиDEFERREDrQzL SELECT COUNT(*) FROM queue WHERE status = 'rurN)rKrBrrrarr8rls rqsizezSQLiteQueueBroker.qsizesd   J  / (4\\'!!+!3!3 45 #F??$Q'  ( ( (s 6AAc|jd5}|jdtjd}t |j d cdddS#1swYyxYw)u2Проверить, пуста ли очередьrrQzm SELECT EXISTS ( SELECT 1 FROM queue WHERE status = 'z ' ) rN)rKrBrrboolrars remptyzSQLiteQueueBroker.emptysn   J  / 24\\'%&0%7%7$89 #F FOO-a011 2 2 2s AAA&c8 |jd5}|jdtjdtjd}t |j d}|s dddy dddtj|#1swYxYw) u Блокирует выполнение, пока в очереди не останется активных элементов. Активные элементы = статус 'pending' ИЛИ 'in_progress'. TrrQz{ SELECT EXISTS ( SELECT 1 FROM queue WHERE status IN ('', 'z)') ) rN) rKrBrrrrrarVre)rcheck_intervalr8rlhas_active_itemss rwaitzSQLiteQueueBroker.waits "" "3 t++,6+=+=*>d:CYCYBZ[' $((9!(<#= ' (  JJ~ & s ABBc|dnd}||fnd}|jd5}d|d}|j||}|jDcic] }|d |d  }}|d } |f} nd } d} |j| | } | j} | d | d nd} | d}| d}| d}d}d}|dkDr||||z }|dkDr||z }n t |}dddt Dcic])}|j j|j d+}}t|j|d< |d<|d<|d<|Scc}w#1swYsxYwcc}w)uПолная статистика по статусам, времени и скорости обработки элементовNz WHERE pid = ?rrrQzu SELECT status, COUNT(*) as count FROM queue z- GROUP BY status statuscountao 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_finishedrtotalavg_execution_time_sectotal_execution_time_secitems_per_second) rKrBfetchallrafloatrrrqsumvalues)rr pid_condition pid_paramr8 query_statusrlrmstats query_metrics metrics_paramcursor_metricsmetricsavg_execution_timerrrtotal_execution_timerrreports r get_statszSQLiteQueueBroker.get_statss+.? !oSF2   J  /1 >4 *?+ L\\, :F2L#& " "{'>fJTTv&,, &,, ::TTfmmo.w+='(-A)*%5!" cO1 >1 >fUs$+E D;A3E.E ;EE c |jd5}|jdtjdtjd}|j Dcic]M}|d|d|d|dnd t j|d |d t j|d nd d Oc}cd d d Scc}w#1swYy xYw)uiВыгрузить результаты всех успешно выполненных элементовrrQz SELECT id, status, payload, result, finished_at - started_at as duration FROM queue WHERE status IN ('rz') r^rdurationNrrYrv)r duration_secrYrv)rKrBrrr rrWrc)rr8rlrms rget_all_resultsz!SQLiteQueueBroker.get_all_resultsOs   J  / 4\\'#$.#7#7"8Z=N=Nx=;TdjjX7Z^        sAB=AB8,B=8B==Cct|jdrS|jj< |jjjt |jdyyy#t$rY$wxYw)umЗакрывает соединение текущего потока и удаляет его из кэша.r8N)r<r-r8rNrHdelattrrs rr1z"SQLiteQueueBroker.close_local_connasg 4;; 'DKK,<,<,H    &&( DKK ( -I '  s$A++ A76A7ct|jdrv|jj` |jjjd|jjj t |jd|j|jj|jjd|jj|jjdg}|D]N}tdD]>}|js# |jtjd|NPy#t $r9 |jjj n#t $rYnwxYwYwxYw#tt f$r6|dk(rtj#d |nt%j&d YwxYw) uaПолное уничтожение очереди и удаление файлов с дискаr8Nz PRAGMA wal_checkpoint(TRUNCATE);z-walz-shmu5Файл очереди успешно удален: uYНе удалось удалить файл очереди после 15 попыток: g?)r<r-r8rBrNrHrr* with_namer&rangeexistsunlinkr2rdPermissionErrorOSErrorrzrVre)rfiles_to_remove file_pathattempts rcleanupzSQLiteQueueBroker.cleanupjs 4;; 'DKK,<,<,H    (()KL   &&( DKK ( MM MM # #t}}'9'9&:$$? @ MM # #t}}'9'9&:$$? @  ) (I 9 ( '')($$&LL#XYbXc!de ( ( KK$$**,  ,(1("} (ABKAL&MN 3 (sIA E"(F F$E?>F? F F F  FFAGG)queueN)r)TNr)g?)r r r r'r6rDrr?rKr.r\rqrxrrrrrrrr1rrrrr"r"!spLI  $$4 47l ... ( 2'&?B$)!(rr")r/rWloggingr_rfr=r+rVrT contextlibrenumrpathlibr getLoggerr2rHrr?rr"rrrrsa   %   > * I dj(j(r