U ڲg&@s0dZddlmZddlmZmZddlmZddlm Z m Z m Z m Z ddl mZddl mZdd lmZdd lmZmZmZmZdd lmZmZdd lmZmZmZmZe rdd lm Z ddl!m"Z"ddl#m$Z$ej%j&Z&ej%j'Z'ej(j)Z)GdddedZ*GdddeZ+Gddde+Z,Gddde,Z-dS)z6 Objects to support the COPY protocol (sync version). ) annotations)ABCabstractmethod) TracebackType)AnyIteratorSequence TYPE_CHECKING)pq)errors)Self)BaseCopyMAX_BUFFER_SIZE QUEUE_SIZE PREFER_FLUSH)copy_tocopy_end)spawngatherQueueWorker)Buffer)Cursor) ConnectioncseZdZUdZdZded<dddddd d fd d Zd dddZdddddddZddddZddddZ dddd Z d!dd"d#Z d$dd%d&d'Z d(dd)d*d+Z ddd,d-d.ZZS)/CopyajManage an asynchronous :sql:`COPY` operation. :param cursor: the cursor where the operation is performed. :param binary: if `!True`, write binary format. :param writer: the object to write to destination. If not specified, write to the `!cursor` connection. Choosing `!binary` is not necessary if the cursor has executed a :sql:`COPY` operation, because the operation result describes the format too. The parameter is useful when a `!Copy` object is created manually and no operation is performed on the cursor, such as when using ``writer=``\ `~psycopg.copy.FileWriter`. ZpsycopgWriterwriterN)binaryr Cursor[Any]z bool | Nonez Writer | None)cursorrrcs.tj||d|st|}||_|j|_dS)N)r)super__init__ LibpqWriterrwrite_write)selfr rr __class__1/tmp/pip-unpacked-wheel-b_ea6rx_/psycopg/_copy.pyr"5s z Copy.__init__r returncCs ||SN)_enterr&r)r)r* __enter__CszCopy.__enter__ztype[BaseException] | NoneBaseException | NonezTracebackType | NoneNone)exc_typeexc_valexc_tbr,cCs||dSr-)finish)r&r3r4r5r)r)r*__exit__Gsz Copy.__exit__zIterator[Buffer]ccs|}|sq|VqdS)z5Implement block-by-block iteration on :sql:`COPY TO`.N)readr&datar)r)r*__iter__Qsz Copy.__iter__rcCs|j|S)z Read an unparsed row after a :sql:`COPY TO` operation. Return an empty string when the data is finished. ) connectionwaitZ _read_genr/r)r)r*r8Ysz Copy.readzIterator[tuple[Any, ...]]ccs|}|dkrq|VqdS)z Iterate on the result of a :sql:`COPY TO` operation record by record. Note that the records returned will be tuples of unparsed strings or bytes, unless data types are specified using `set_types()`. N)read_row)r&recordr)r)r*rowsasz Copy.rowsztuple[Any, ...] | NonecCs|j|S)a Read a parsed row of data from a table after a :sql:`COPY TO` operation. Return `!None` when the data is finished. Note that the records returned will be tuples of unparsed strings or bytes, unless data types are specified using `set_types()`. )r<r=Z _read_row_genr/r)r)r*r>ns z Copy.read_rowz Buffer | str)bufferr,cCs|j|}|r||dS)z Write a block of data to a table after a :sql:`COPY FROM` operation. If the :sql:`COPY` is in binary format `!buffer` must be `!bytes`. In text mode it can be either `!bytes` or `!str`. N) formatterr$r%)r&rAr:r)r)r*r$ys z Copy.writez Sequence[Any])rowr,cCs|j|}|r||dS)z=Write a record to a table after a :sql:`COPY FROM` operation.N)rB write_rowr%)r&rCr:r)r)r*rDs zCopy.write_rowexcr,cCsl|jtkr6|j}|r"|||j|d|_n2|s>dS|jj t krNdS|j |j |dS)aTerminate the copy operation and free the resources allocated. You shouldn't need to call this function yourself: it is usually called by exit. It is available if, despite what is documented, you end up using the `Copy` object outside a block. TN)Z _directionCOPY_INrBendr%rr6 _finished_pgconnZtransaction_statusACTIVEr<Z _try_cancelr=Z_end_copy_out_gen)r&rFr:r)r)r*r6s      z Copy.finish)__name__ __module__ __qualname____doc____annotations__r"r0r7r;r8r@r>r$rDr6 __classcell__r)r)r'r*r"s     rzConnection[Any]c@s6eZdZdZedddddZd ddd d d ZdS) rzG A class to write copy data somewhere (for async connections). rr2r:r,cCsdS)zWrite some data to destination.Nr)r9r)r)r*r$sz Writer.writeNr1rEcCsdS)z Called when write operations are finished. If operations finished with an error, it will be passed to ``exc``. Nr)r&rFr)r)r*r6sz Writer.finish)N)rLrMrNrOrr$r6r)r)r)r*rsrc@sDeZdZdZdZddddZddd d d Zdd ddddZd S)r#z@ An `Writer` to write copy data to a Postgres database. psycopg.copyrr cCs||_|j|_|jj|_dSr-)r r<ZpgconnrJr&r r)r)r*r"szLibpqWriter.__init__rr2rRc Csdt|tkr&|jt|j|tdn:tdt|tD](}|jt|j|||ttdq6dS)Nflushr)lenrr<r=rrJrranger&r:ir)r)r*r$s zLibpqWriter.writeNr1rEcCst|r,dt|jd|}||jjd}nd}z|jt|j|}Wntj k rd|s`Yn X|g|j _ dS)Nzerror from Python: z - replace) typerNencoderJ _encodingr<r=reZ QueryCanceledr Z_results)r&rFmsgZbmsgresr)r)r*r6szLibpqWriter.finish)N)rLrMrNrOr"r$r6r)r)r)r*r#s r#cs^eZdZdZdZddfdd Zddd d Zd dd d dZddddfdd ZZS)QueuedLibpqWriterz `Writer` using a buffer to queue data to write. `write()` returns immediately, so that the main thread can be CPU-bound formatting messages, while a worker thread can be IO-bound waiting to write on the connection. rTrrUcs(t|ttd|_d|_d|_dS)N)maxsize)r!r"rr_queue_worker _worker_errorrVr'r)r*r"s  zQueuedLibpqWriter.__init__r2r+c Cs\z.|j}|sq,|jt|j|tdqWn(tk rV}z ||_W5d}~XYnXdS)zPush data to the server when available from the copy queue. Terminate reading when the queue receives a false-y value, or in case of error. The function is designed to be run in a separate task. rWN) rfgetr<r=rrJr BaseExceptionrh)r&r:exr)r)r*workers zQueuedLibpqWriter.workerrrRcCsj|jst|j|_|jr|jt|tkr8|j|n.tdt|tD]}|j|||tqHdS)Nr) rgrrlrhrYrrfputrZr[r)r)r*r$ s  zQueuedLibpqWriter.writeNr1rEcs>|jd|jr"t|jd|_|jr.|jt|dS)N)rfrmrgrrhr!r6rSr'r)r*r6s  zQueuedLibpqWriter.finish)N) rLrMrNrOr"rlr$r6rQr)r)r'r*rds rdN).rO __future__rabcrrtypesrtypingrrrr r r ra_compatr Z _copy_baserrrr generatorsrrZ_acompatrrrrrr rr<rZ ExecStatusrGZCOPY_OUTZTransactionStatusrKrrr#rdr)r)r)r*s,        /