U
S��W[! � @ s d dl Z d dlZd dlmZ d dlmZ zd dlmZ W n ek
rX d dlmZ Y nX z0d dlZd dlm Z
d dl mZ d dl
mZ W n ek
r� d Z
ZZ Y nX d dlmZ e �d�ZG dd � d e�ZG d
d � d e�Zd Zd
ZG dd� de�ZG dd� de�ZG dd� de�ZdS )� N��Event)�Thread)�Queue)�Greenlet)�SqliteExtDatabasezpeewee.sqliteqc @ s e Zd ZdS )�
ResultTimeoutN)�__name__�
__module__� __qualname__� r r � /sqliteq.pyr s r c @ s| e Zd ZdZdd� Zddd�Zddd�Zd d
� Zd d � ZeZ e
d
d� �Z e
dd� �Z e
dd� �Z
dd� Zdd� Zdd� ZdS )� AsyncCursor) �sql�params�commit�timeout�_event�_cursor�_exc�_idx�_rowsc C s: || _ || _|| _|| _|| _d | _ | _ | _| _d S �N) r r r r r r r r r )�self�eventr r r r r r r
�__init__ s zAsyncCursor.__init__Nc C s6 || _ || _d| _|d kr"|�� ng | _| j�� | S )Nr )r r r �fetchallr r �set)r �cursor�excr r r
�
set_result( s
zAsyncCursor.set_resultc C s@ |d k r |n| j }| jj|d�s,|r,td��| jd k r<| j�d S )N)r z results not ready, timed out.)r r �waitr r )r r r r r
�_wait0 s
zAsyncCursor._waitc C s | � � | jd k r| j�| S r )r" r �_exec�r r r r
�__iter__7 s
zAsyncCursor.__iter__c C s@ z| j | j }W n tk
r( t�Y nX | jd7 _|S d S )N� )r r �
IndexError�
StopIteration)r �objr r r
�next= s
zAsyncCursor.nextc C s | � � | jjS r )r" r � lastrowidr$ r r r
r+ G s zAsyncCursor.lastrowidc C s | � � | jjS r )r" r �rowcountr$ r r r
r, L s zAsyncCursor.rowcountc C s | j jS r )r � descriptionr$ r r r
r- Q s zAsyncCursor.descriptionc C s | j �� d S r )r �closer$ r r r
r. U s zAsyncCursor.closec C s t | �S r )�listr$ r r r
r X s zAsyncCursor.fetchallc C s. | � � z
t| �W S tk
r( Y d S X d S r )r" r* r( r$ r r r
�fetchone[ s
zAsyncCursor.fetchone)N)N)r r
r � __slots__r r r" r% r* �__next__�propertyr+ r, r- r. r r0 r r r r
r s
r z�threadlocals cannot be set to True when using the Sqlite thread / queue database. All queries are serialized through a single connection, so allowing multiple threads to connect defeats the purpose of this database.z�SQLite must be configured to use the WAL journal mode when using this feature. WAL mode allows one or more readers to continue reading while another connection writes to the database.c sr e Zd Zd� fdd� Zdd� Zd dd �Zd
d � Zd d
� Zdd� Zdd� Z d dd�Z
dd� Z dd� Z dd� Z
� ZS ) �SqliteQueueDatabaseFr& Nc s� |� d�rtt��d|d< d|d< | �|�dd �|�dd ��} tt| �}
|
j| _|
j |f|�d| i|�� || _
|| _ || _ d| _
| �|�|�| _| �� | j
r�| �� d S )NZ threadlocalsFZcheck_same_thread� journal_mode�pragmasT)�get�
ValueError�THREADLOCAL_ERROR_MESSAGE�_validate_journal_mode�pop�superr4 � execute_sql�!_SqliteQueueDatabase__execute_sqlr Z
_autostart�_results_timeout� _num_readers� _is_stopped�get_thread_impl�_thread_helper�_create_queues_and_workers�start) r Zdatabase�
use_geventZ autostartZreaders�queue_max_sizeZresults_timeout�args�kwargsr6 ZParent�� __class__r r
r o s&
�
z SqliteQueueDatabase.__init__c C s |rt S tS r )�GreenletHelper� ThreadHelper)r rF r r r
rB � s z#SqliteQueueDatabase.get_thread_implc C sh |r|� � dkrtt��|r^tdd� |D ��}|�dd�� � dkrJtt��dd� |D �dg S dgS d S )N�walc s s | ]\}}|� � |fV qd S r )�lower��.0�k�vr r r
� <genexpr>� s z=SqliteQueueDatabase._validate_journal_mode.<locals>.<genexpr>r5 c S s g | ]\}}|d kr||f�qS )r5 r rP r r r
�
<listcomp>� s �z>SqliteQueueDatabase._validate_journal_mode.<locals>.<listcomp>)r5 rN )rO r8 �WAL_MODE_ERROR_MESSAGE�dictr7 )r r5 r6 Zpdictr r r
r: � s �z*SqliteQueueDatabase._validate_journal_modec sP � j �� � _� j �� � _� j�� j ��� j�� _� �fdd�t� j�D �� _ d S )Nc s g | ]}� j ��� j��qS r )rC �thread� _read_queue)rQ �_�r �targetr r
rU � s �zBSqliteQueueDatabase._create_queues_and_workers.<locals>.<listcomp>)
rC �queue� _write_queuerY �_run_worker_looprX �_writer�ranger@ �_readersr$ r r[ r
rD � s �z.SqliteQueueDatabase._create_queues_and_workersc C s< |� � }|tkr t�d� d S t�d|j� | �|� q d S )Nzworker shutting down.zreceived query %s)r7 r( �logger�info�debugr �_process_execution)r r] � async_cursorr r r
r_ � s
z$SqliteQueueDatabase._run_worker_loopc
C sP z| � |j|j|j�}W n& tk
r> } zd }W 5 d }~X Y nX d }|�||�S r )r> r r r � Exceptionr )r rg r r r r r
rf � s �z&SqliteQueueDatabase._process_executionc C s | j �� | j�� fS r )r^ ZqsizerY r$ r r r
�
queue_size� s z SqliteQueueDatabase.queue_sizeTc C sD t | j�� ||||d kr | jn|d�}|r0| jn| j}|�|� |S )N)r r r r r )r rC r r? r^ rY �put)r r r Zrequire_commitr r r] r r r
r= � s �
z SqliteQueueDatabase.execute_sqlc C sb | j �R | js W 5 Q R � dS | j�� | jD ] }|�� q,t�d� d| _W 5 Q R � dS Q R X d S )NFzworkers started.T)�
_conn_lockrA r` rE rb rc rd )r �readerr r r
rE � s
zSqliteQueueDatabase.startc C s� t �d� | j�f | jr&W 5 Q R � dS | j�t� | jD ]}| j�t� q8| j �
� | jD ] }|�
� qZW 5 Q R � dS Q R X d S )Nzenvironment stop requested.FT) rc re rk rA r^ rj r( rb rY r` �join)r rZ rl r r r
�stop� s
zSqliteQueueDatabase.stopc
C s$ | j � | jW 5 Q R � S Q R X d S r )rk rA r$ r r r
�
is_stopped� s z SqliteQueueDatabase.is_stopped)FFr& NN)NN)NTN)r r
r r rB r: rD r_ rf ri r= rE rn ro �
__classcell__r r rJ r
r4 n s �!
r4 c @ s4 e Zd ZdZd dd�Zdd� Zd dd�Zd d
� ZdS )
rM �rG Nc C s
|| _ d S r rq )r rG r r r
r � s zThreadHelper.__init__c C s t � S r r r$ r r r
r � � zThreadHelper.eventc C s |d k r |n| j }t|pdd�S �Nr )�maxsize)rG r �r Zmax_sizer r r
r] � s zThreadHelper.queuec O s t |||d�}d|_|S )N)r\ rH rI T)r Zdaemon)r �fnrH rI rX r r r
rX � s zThreadHelper.thread)N)N)r r
r r1 r r r] rX r r r r
rM � s
rM c @ s* e Zd ZdZdd� Zd dd�Zdd� ZdS )
rL rq c C s t � S r )�GEventr$ r r r
r � rr zGreenletHelper.eventNc C s |d k r |n| j }t|pdd�S rs )rG �GQueueru r r r
r] s zGreenletHelper.queuec s � fdd�}t |f|�|�S )Nc s t �� � | |�S r )�gevent�sleep)�arR �rv r r
�wrap s z#GreenletHelper.thread.<locals>.wrap)�GThread)r rv rH rI r} r r| r
rX s zGreenletHelper.thread)N)r r
r r1 r r] rX r r r r
rL � s
rL ) Zlogging�weakrefZ threadingr r r � ImportErrorr] ry r r~ Z gevent.eventrw Z gevent.queuerx Zplayhouse.sqlite_extr Z getLoggerrc rh r �objectr r9 rV r4 rM rL r r r r
�<module> s.
G{