golden hour
/opt/saltstack/salt/lib/python3.10/site-packages/salt/queues/__pycache__
⬆️ Go Up
Upload
File/Folder
Size
Actions
__init__.cpython-310.pyc
195 B
Del
OK
pgjsonb_queue.cpython-310.pyc
7.28 KB
Del
OK
sqlite_queue.cpython-310.pyc
6.78 KB
Del
OK
Edit: pgjsonb_queue.cpython-310.pyc
o �xe� � @ s� d Z ddlZddlZddlmZ ddlZddlmZm Z zddl Z dZ W n e y/ dZ Y nw e� e�ZdZdd � Zed#d d ��Zd d � Zdd� Zdd� Zdd� Zdd� Zdd� Zdd� Zdd� Zd d � Zd d � Z d$d!d"�Z dS )%a� .. versionadded:: 2016.3.0 This is a queue with postgres as the backend. It uses the jsonb store to store information for queues. :depends: python-psycopg2 To enable this queue, the following needs to be configured in your master config. These are the defaults: .. code-block:: yaml queue.pgjsonb.host: 'salt' queue.pgjsonb.user: 'salt' queue.pgjsonb.password: 'salt' queue.pgjsonb.dbname: 'salt' queue.pgjsonb.port: 5432 Use the following Pg database schema: .. code-block:: sql CREATE DATABASE salt WITH ENCODING 'utf-8'; -- -- Table structure for table `salt` -- DROP TABLE IF EXISTS salt; CREATE OR REPLACE TABLE salt( id SERIAL PRIMARY KEY, data jsonb NOT NULL ); .. code-block:: bash salt-run queue.insert test '{"name": "redis", "host": "172.16.0.8", "port": 6379}' backend=pgjsonb salt-run queue.process_queue test all backend=pgjsonb � N)�contextmanager)�SaltInvocationError�SaltMasterErrorTFZpgjsonbc C s t du rdS tS )NF)�HAS_PG�__virtualname__� r r �M/opt/saltstack/salt/lib/python3.10/site-packages/salt/queues/pgjsonb_queue.py� __virtual__? s r c c s � dddddd�}i }|� � D ]\}}t�d�t|�|�||< qz tjd i |��}W n tjy? } ztdj|d���d}~ww |� � }z?z|V W n tj yh } z|j } t j �t| �� |�d � � d}~ww | rq|�d � n |�d � W |�� dS W |�� dS |�� w ) z# Return an postgres cursor Z localhost�salti8 )�host�user�passwordZdbname�portz queue.{}.{}z5pgjsonb returner could not connect to database: {exc})�excNZROLLBACKZCOMMITr )�itemsZ__opts__�get�formatr �psycopg2�connectZOperationalErrorr �cursorZ DatabaseError�args�sys�stderr�write�str�execute�close) �commit�defaultsZ conn_kwargs�key�value�connr r �err�errorr r r �_connE sH �� � ��� �� �r$ c C s0 d}t �d|� | �|� | �� }dd� |D �S )NzLselect relname from pg_class where relkind='r' and relname !~ '^(pg_|sql_)';� SQL Query: %sc S � g | ]}|d �qS �r r )�.0�xr r r � <listcomp>u � z _list_tables.<locals>.<listcomp>)�log�debugr �fetchall)�cur�cmd�resultr r r � _list_tablesp s r2 c C s$ d� |�}t�d|� | �|� dS )Nz;CREATE TABLE {}(id SERIAL PRIMARY KEY, data jsonb NOT NULL)r% T)r r, r- r )r/ �queuer0 r r r � _create_tablex s r4 c C sV t � � }d�| �}t�d|� |�|� |�� }|W d � S 1 s$w Y dS )z6 Private function to list contents of a queue zSELECT data FROM {}r% N)r$ r r, r- r r. )r3 r/ r0 �contentsr r r � _list_items s $�r6 c C s4 t � � } t| �}W d � |S 1 sw Y |S )z9 Return a list of Salt Queues on the Salt Master N)r$ r2 )r/ Zqueuesr r r � list_queues� s ��r7 c C s t | �}dd� |D �}|S )z" List contents of a queue c S r& r' r �r( �itemr r r r* � r+ z list_items.<locals>.<listcomp>)r6 )r3 Z itemstupler r r r � list_items� s r: c C s t | �}t|�S )z0 Provide the number of items in a queue )r6 �len)r3 r r r r � list_length� s r< c C s | t � v S )z� Does this queue exist :param queue: Name of the queue :type str :return: True if this queue exists and False otherwise :rtype bool )r7 )r3 r r r � _queue_exists� s r= c C s^ t | �s'tdd��}t�d| � t|| � W d � d S 1 s w Y d S t�d| � d S )NT�r z!Queue %s does not exist. CreatingzQueue %s already exists.)r= r$ r, r- r4 )r3 r/ r r r �handle_queue_creation� s "�r? c C sD t | � tdd���}t|t�rItjj�|�}d�| |�}t � d|� z|� |� W n t j yH } zd�|�W Y d}~W d � S d}~ww t|t�r�dd� |D �}d �| �}t � d|� z|�||� W n' t j y� } zd �|�W Y d}~W d � S d}~ww W d � dS W d � dS 1 s�w Y dS ) z) Add an item or items to a queue Tr> z!INSERT INTO {}(data) VALUES('{}')r% z5Item already exists in this queue. postgres error: {}Nc S � g | ] }t jj�|�f�qS r �r �utils�json�dumps�r( �elr r r r* � � zinsert.<locals>.<listcomp>z INSERT INTO {}(data) VALUES (%s)zBOne or more items already exists in this queue. postgres error: {})r? r$ � isinstance�dictr rB rC rD r r, r- r r ZIntegrityError�list� executemany)r3 r r/ r0 �escr r r �insert� sH ��� ���� �� ��rM c C s� t dd��R}t|t�r*d�| tjj�|��}t� d|� |� |� W d � dS t|t �rOdd� |D �}d�| �}t� d|� |� ||� W d � dS W d � dS 1 sZw Y dS ) z. Delete an item or items from a queue Tr> z DELETE FROM {} WHERE data = '{}'r% Nc S r@ r rA rE r r r r* � rG zdelete.<locals>.<listcomp>z DELETE FROM {} WHERE data = %s) r$ rH rI r r rB rC rD r, r- r rJ rK )r3 r r/ r0 r r r �delete� s* � � � � � � rN � c C s d� | �}|dkr/zt|�}W n ty$ } z d� |�}t|��d}~ww d�|d� |�g�}t�d|� g }tdd ��D}|�|� |� � }t |�d krxd d � |D �} d d � |D �}d�| �} d� | | �} t�d| � |�| � W d � |S W d � |S 1 s�w Y |S )zB Pop one or more or all items from the queue return them. zSELECT id, data FROM {}�allz2Quantity must be an integer or "all". Error: "{}".N� z LIMIT {};r% Tr> r c S s g | ]}t |d ��qS r' )r r8 r r r r* s zpop.<locals>.<listcomp>c S r& )rO r r8 r r r r* r+ z','z"DELETE FROM {} WHERE id IN ('{}');) r �int� ValueErrorr �joinr, r- r$ r r. r; ) r3 ZquantityZ is_runnerr0 r Z error_txtr r/ r1 ZidsZidlistZdel_cmdr r r �pop� s> ��� � � � � rU )F)rO F) �__doc__Zloggingr � contextlibr Zsalt.utils.jsonr Zsalt.exceptionsr r r r � ImportErrorZ getLogger�__name__r, r r r$ r2 r4 r6 r7 r: r< r= r? rM rN rU r r r r �<module> s8 ) � *
Save