j&dZddlmZddlmZmZddlmZddlm Z m Z m Z m Z ddl mZddl mZdd lmZdd lmZmZmZmZdd lmZmZmZmZdd lmZmZe rdd lm Z ddl!m"Z"ddl#m$Z$ejJjLZ&ejJjNZ'ejPjRZ)GddedZ*GddeZ+Gdde+Z,Gdde,Z-y)z7 Objects to support the COPY protocol (async version). ) annotations)ABCabstractmethod) TracebackType) TYPE_CHECKINGAny AsyncIteratorSequence)errors)pq)Self)AQueueAWorkeragatheraspawn)MAX_BUFFER_SIZE PREFER_FLUSH QUEUE_SIZEBaseCopy)copy_endcopy_to)Buffer) AsyncCursor)AsyncConnectionceZdZUdZdZded<ddd dfdZddZ dd Zdd Zdd Z dd Z dd Z ddZ ddZ ddZxZS) AsyncCopyajManage 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`. psycopg AsyncWriterwriterN)binaryr crt||||s t|}||_|j|_y)N)r!)super__init__AsyncLibpqWriterr write_write)selfcursorr!r __class__s Z/var/www/html/kelly/kelly-backend/venv/lib/python3.12/site-packages/psycopg/_copy_async.pyr$zAsyncCopy.__init__2s6 /%f-F ll c.K|j|SwN)_enterr(s r+ __aenter__zAsyncCopy.__aenter__@s  scBK|j|d{y7wr.)finish)r(exc_typeexc_valexc_tbs r+ __aexit__zAsyncCopy.__aexit__Ds kk'"""s cK|jd{x}r!||jd{x}r yy7(7 w)z5Implement block-by-block iteration on :sql:`COPY TO`.N)readr(datas r+ __aiter__zAsyncCopy.__aiter__Ns<!YY[()d)J"YY[()d)((#A?AAAAAcpK|jj|jd{S7w)z Read an unparsed row after a :sql:`COPY TO` operation. Return an empty string when the data is finished. N) connectionwait _read_genr0s r+r9zAsyncCopy.readSs* __))$..*:;;;; -646cK|jd{x}!||jd{x} yy7(7 w)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(records r+rowszAsyncCopy.rows[s>!% /0v=L!% /0v=//r=cpK|jj|jd{S7w)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()`. N)r?r@ _read_row_genr0s r+rDzAsyncCopy.read_rowes,__))$*<*<*>????rBc~K|jj|x}r|j|d{yy7w)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(bufferr;s r+r&zAsyncCopy.writeps<>>''/ /4 /++d# # # 0 # 2=;=c~K|jj|x}r|j|d{yy7w)z=Write a record to a table after a :sql:`COPY FROM` operation.N)rJ write_rowr')r(rowr;s r+rNzAsyncCopy.write_rowzs:>>++C0 04 0++d# # # 1 #rLcK|jtk(rb|s5|jjx}r|j |d{|j j |d{d|_y|sy|jjtk7ry|jjd{|jj|jd{y777:7 w)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. NT) _directionCOPY_INrJendr'r r3 _finished_pgconntransaction_statusACTIVEr? _try_cancelr@_end_copy_out_gen)r(excr;s r+r3zAsyncCopy.finishs ??g %>>--//4/++d+++++$$S) ) )!DN||..&8 //--/ / ///&&t'='='?@ @ @#, ) 0 @sIAC5C- "C5+C/,A C56C170C5'C3(C5/C51C53C5)r)AsyncCursor[Any]r!z bool | Noner zAsyncWriter | None)returnr)r4ztype[BaseException] | Noner5BaseException | Noner6zTracebackType | Noner\None)r\zAsyncIterator[Buffer])r\r)r\zAsyncIterator[tuple[Any, ...]])r\ztuple[Any, ...] | None)rKz Buffer | strr\r^)rOz Sequence[Any]r\r^rZr]r\r^)__name__ __module__ __qualname____doc____annotations__r$r1r7r<r9rFrDr&rNr3 __classcell__r*s@r+rrs J  #%) #  # # # ##,#&#% #  # < @$$ Ar,rzAsyncConnection[Any]c,eZdZdZeddZdddZy)rzG A class to write copy data somewhere (for async connections). c Kyw)zWrite some data to destination.Nr:s r+r&zAsyncWriter.writes  Nc Kyw)z Called when write operations are finished. If operations finished with an error, it will be passed to ``exc``. Nri)r(rZs r+r3zAsyncWriter.finishs rjr;rr\r^r.r_)r`rarbrcrr&r3rir,r+rrs    r,rc.eZdZdZdZddZddZd d dZy) r%zE An `AsyncWriter` to write copy data to a Postgres database. psycopg.copycj||_|j|_|jj|_yr.)r)r?pgconnrU)r(r)s r+r$zAsyncLibpqWriter.__init__s'  ++-- r,c Kt|tkr>|jjt |j |t d{ytdt|tD]I}|jjt |j |||tzt d{Ky7l7 w)Nflushr)lenrr?r@rrUrranger(r;is r+r&zAsyncLibpqWriter.writes t9 '//&&wt||T'VW W W1c$i9oo** d1q?/B&C<: X s%A B> B:A$B>2B<3B><B>NchK|rBdt|jd|}|j|jjd}nd} |j j t|j|d{}|g|j_ y7#tj$r|sYywxYww)Nzerror from Python: z - replace) typerbencoderU _encodingr?r@rr)_resultse QueryCanceled)r(rZmsgbmsgress r+r3zAsyncLibpqWriter.finishs 'S (>(>'?s3%HC::dll44i@DD ),,XdllD-IJJC%(5DKK K  s<AB2 2B<B=BB2BB/,B2.B//B2r)r[rlr.r_)r`rarbrcr$r&r3rir,r+r%r%s J. )r,r%cFeZdZdZdZdfd ZddZd dZd d fd ZxZS) AsyncQueuedLibpqWriterz `AsyncWriter` 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. rncjt||tt|_d|_d|_y)N)maxsize)r#r$rr_queue_worker _worker_error)r(r)r*s r+r$zAsyncQueuedLibpqWriter.__init__s+  &,Z&@ '+ 37r,cbK |jjd{x}rc|jjt |j |t d{|jjd{x}rbyy7j7-7 #t$r}||_Yd}~yd}~wwxYww)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. Nrr) rgetr?r@rrUr BaseExceptionr)r(r;exs r+workerzAsyncQueuedLibpqWriter.workers $!%!223$3oo**DLL$lC"&!223$323 $!#D   $sbB/BB >B B!!BBB B/ BBB B,B'"B/'B,,B/cK|jst|j|_|jr |jt |t kr$|j j|d{ytdt |t D]/}|j j|||t zd{1y7R7 w)Nr) rrrrrtrrputrurvs r+r&zAsyncQueuedLibpqWriter.writes||!$++.DL   $$ $ t9 '++//$' ' '1c$i9kkood1q?/B&CDDD: ( Es%A/C1C2A C<C=CCcK|jjdd{|jr$t|jd{d|_|jr |jt ||d{y7d7=7 w)Nr,)rrrrrr#r3)r(rZr*s r+r3zAsyncQueuedLibpqWriter.finishswkkooc""" <<$,,' ' 'DL   $$ $gnS!!! # ( "s3B B(B  B  5B B B  B  B r)r\r^rlr.r_) r`rarbrcr$rr&r3rerfs@r+rrs) J8$"E& " "r,rN).rc __future__rabcrrtypesrtypingrrr r r r~r _compatr_acompatrrrr _copy_baserrrr generatorsrrr cursor_asyncrconnection_asyncr ExecStatusrRCOPY_OUTTransactionStatusrWrrr%rrir,r+rs ##>>66KK))1 --   == ! !   $ ${A/0{A| # &,){,)^A"-A"r,