U SW[!@sddlZddlZddlmZddlmZzddlmZWn ek rXddlmZYnXz0ddlZddlm Z ddl mZ ddl mZWn ek rdZ ZZ YnXddlmZedZGdd d eZGd d d eZd Zd ZGdddeZGdddeZGdddeZdS)NEvent)Thread)Queue)Greenlet)SqliteExtDatabasezpeewee.sqliteqc@s eZdZdS) ResultTimeoutN)__name__ __module__ __qualname__r r /sqliteq.pyrsrc@s|eZdZdZddZdddZdddZd d Zd d ZeZ e d dZ e ddZ e ddZ ddZddZddZdS) AsyncCursor) sqlparamscommittimeout_event_cursor_exc_idx_rowscCs:||_||_||_||_||_d|_|_|_|_dSN) rrrrrrrrr)selfeventrrrrr r r __init__ s zAsyncCursor.__init__NcCs6||_||_d|_|dkr"|ng|_|j|S)Nr)rrrfetchallrrset)rcursorexcr r r set_result(s  zAsyncCursor.set_resultcCs@|dk r |n|j}|jj|ds,|r,td|jdk r<|jdS)N)rzresults not ready, timed out.)rrwaitrr)rrr r r _wait0s  zAsyncCursor._waitcCs||jdk r|j|Sr)r"r_execrr r r __iter__7s zAsyncCursor.__iter__cCs@z|j|j}Wntk r(tYnX|jd7_|SdS)N)rr IndexError StopIteration)robjr r r next=s  zAsyncCursor.nextcCs||jjSr)r"r lastrowidr$r r r r+GszAsyncCursor.lastrowidcCs||jjSr)r"rrowcountr$r r r r,LszAsyncCursor.rowcountcCs|jjSr)r descriptionr$r r r r-QszAsyncCursor.descriptioncCs|jdSr)rcloser$r r r r.UszAsyncCursor.closecCst|Sr)listr$r r r rXszAsyncCursor.fetchallcCs.|z t|WStk r(YdSXdSr)r"r*r(r$r r r fetchone[s  zAsyncCursor.fetchone)N)N)r r r __slots__rr r"r%r*__next__propertyr+r,r-r.rr0r r r r rs      rzthreadlocals 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.zSQLite 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.csreZdZdfdd ZddZddd Zd d Zd d ZddZddZ dddZ ddZ ddZ ddZ ZS)SqliteQueueDatabaseFr&Nc s|drttd|d<d|d<||dd|dd} tt|} | j|_| j |f|d| i|||_ ||_ ||_ d|_ ||||_||j r|dS)NZ threadlocalsFZcheck_same_thread journal_modepragmasT)get ValueErrorTHREADLOCAL_ERROR_MESSAGE_validate_journal_modepopsuperr4 execute_sql!_SqliteQueueDatabase__execute_sqlrZ _autostart_results_timeout _num_readers _is_stoppedget_thread_impl_thread_helper_create_queues_and_workersstart) rZdatabase use_geventZ autostartZreadersqueue_max_sizeZresults_timeoutargskwargsr6ZParent __class__r r ros&    zSqliteQueueDatabase.__init__cCs |rtStSr)GreenletHelper ThreadHelper)rrFr r r rBsz#SqliteQueueDatabase.get_thread_implcCsh|r|dkrtt|r^tdd|D}|dddkrJttdd|DdgSdgSdS)Nwalcss|]\}}||fVqdSr)lower.0kvr r r sz=SqliteQueueDatabase._validate_journal_mode..r5cSs g|]\}}|dkr||fqS)r5r rPr r r sz>SqliteQueueDatabase._validate_journal_mode..)r5rN)rOr8WAL_MODE_ERROR_MESSAGEdictr7)rr5r6Zpdictr r r r:s z*SqliteQueueDatabase._validate_journal_modecsPj_j_jjj_fddtjD_ dS)Ncsg|]}jjqSr )rCthread _read_queue)rQ_rtargetr r rUszBSqliteQueueDatabase._create_queues_and_workers..) rCqueue _write_queuerY_run_worker_looprX_writerranger@_readersr$r r[r rDs   z.SqliteQueueDatabase._create_queues_and_workerscCs<|}|tkrtddStd|j||qdS)Nzworker shutting down.zreceived query %s)r7r(loggerinfodebugr_process_execution)rr] async_cursorr r r r_s  z$SqliteQueueDatabase._run_worker_loopc CsPz||j|j|j}Wn&tk r>}zd}W5d}~XYnXd}|||Sr)r>rrr Exceptionr )rrgrrr r r rfs z&SqliteQueueDatabase._process_executioncCs|j|jfSr)r^ZqsizerYr$r r r queue_sizeszSqliteQueueDatabase.queue_sizeTcCsDt|j||||dkr|jn|d}|r0|jn|j}|||S)N)rrrrr)rrCrr?r^rYput)rrrZrequire_commitrrr]r r r r=s zSqliteQueueDatabase.execute_sqlc Csb|jR|jsW5QRdS|j|jD] }|q,tdd|_W5QRdSQRXdS)NFzworkers started.T) _conn_lockrAr`rErbrcrd)rreaderr r r rEs    zSqliteQueueDatabase.startc Cstd|jf|jr&W5QRdS|jt|jD]}|jtq8|j |jD] }| qZW5QRdSQRXdS)Nzenvironment stop requested.FT) rcrerkrAr^rjr(rbrYr`join)rrZrlr r r stops      zSqliteQueueDatabase.stopc Cs$|j|jW5QRSQRXdSr)rkrAr$r r r is_stoppedszSqliteQueueDatabase.is_stopped)FFr&NN)NN)NTN)r r r rrBr:rDr_rfrir=rErnro __classcell__r r rJr r4ns!       r4c@s4eZdZdZd ddZddZd ddZd d ZdS) rMrGNcCs ||_dSrrq)rrGr r r rszThreadHelper.__init__cCstSrrr$r r r rzThreadHelper.eventcCs |dk r |n|j}t|pddSNr)maxsize)rGrrZmax_sizer r r r]szThreadHelper.queuecOst|||d}d|_|S)N)r\rHrIT)rZdaemon)rfnrHrIrXr r r rXszThreadHelper.thread)N)N)r r r r1rrr]rXr r r r rMs   rMc@s*eZdZdZddZd ddZddZdS) rLrqcCstSr)GEventr$r r r rrrzGreenletHelper.eventNcCs |dk r |n|j}t|pddSrs)rGGQueuerur r r r]szGreenletHelper.queuecsfdd}t|f||S)Ncst||Sr)geventsleep)arRrvr r wrapsz#GreenletHelper.thread..wrap)GThread)rrvrHrIr}r r|r rXs zGreenletHelper.thread)N)r r r r1rr]rXr r r r rLs rL)ZloggingweakrefZ threadingrrr ImportErrorr]ryrr~Z gevent.eventrwZ gevent.queuerxZplayhouse.sqlite_extrZ getLoggerrcrhrobjectrr9rVr4rMrLr r r r s.      G{