U ڲg E@sdZddlmZddlZddlmZddlmZddlm Z m Z m Z m Z m Z mZddlmZdd lmZdd lmZdd lmZdd lmZmZmZmZmZmZmZdd lmZddl m!Z!m"Z"m#Z#m$Z$ddl%m&Z&ddl'm(Z(ddl)m*Z*ddl+m,Z,m-Z-ddl+m.Z.m/Z/ddl0m1Z1ddl2m3Z3ddl4m5Z5ddl6m7Z7ddl8m9Z9ddl:m;Z;ddlZ>m?Z?ddl@Z@ddlAZAddlAmBZBddl)mCZCerddlDmEZEdZFejGjHZHejGjIZIejJjKZKejJjLZLejJjMZMeAjNeOfZPeQdZRGd d!d!e=e!ZSdS)"z+ Psycopg connection object (async version) ) annotationsN) monotonic) TracebackType)AnyAsyncGenerator AsyncIteratorcastoverload TYPE_CHECKING)asynccontextmanager)pq)errors)waiting) AdaptContextConnDict ConnParamParamsPQGenQueryRV)Xid)RowAsyncRowFactory tuple_rowargs_row) AdaptersMap)IsolationLevel)Self) make_conninfoconninfo_to_dict)conninfo_attempts_asynctimeout_from_conninfo) AsyncPipeline)notifies)AsyncTransaction) AsyncCursor) capabilities)AsyncServerCursor)BaseConnection CursorRowNotify)Lock) to_thread)PGconng?psycopgc seZdZUdZdZded<ded<ded<d ed <eeeefd dd fd d Z e ddddddddddddddddddZ ddd d!Z d"d#d$d%d&d'd(Z e dd)d*d+d,d-Zd%dd.d/Zedd0dd1d2d3d4Zedd0dd5d6d7d8d4Zedddd9ddd:dd;d<d=d4Zedddd9ddd5d:dd>d?d@d4ZddddddAdddBd:ddCd?dDd4ZddddEdFdGd:dd1dHdIdJZd%ddKdLZd%ddMdNZdOdPdQd%dRdSdTZdUdPdQd%dRdVdWZeddXddYdZd[d\Zddd]d^dd_d`dadbZedcddddeZefdfd^dgdhdidjZdd%dkdldmZdd%dkdndoZdpd%dkdqdrZdpd%dkdsdtZ d:d%dkdudvZ!d:d%dkdwdxZ"d:d%dkdydzZ#d:d%dkd{d|Z$dd%d}d~dZ%dd%dddZ&d%dddZ'ddd%dddZ(ddd%dddZ)ddddZ*Z+S)AsyncConnectionz3 Wrapper for a connection to the database. r/ztype[AsyncCursor[Row]]cursor_factoryztype[AsyncServerCursor[Row]]server_cursor_factoryzAsyncRowFactory[Row] row_factoryzAsyncPipeline | None _pipeliner.)pgconnr3cs*t|||_t|_t|_t|_dSN) super__init__r3r,lockr&r1r(r2)selfr5r3 __class__connectUsL        zAsyncConnection.connect)rHcs|Sr6r=r:r=r=r> __aenter__szAsyncConnection.__aenter__ztype[BaseException] | NonezBaseException | NonezTracebackType | NoneNone)exc_typeexc_valexc_tbrHc s|jr dS|rTz|IdHWqbtk rP}ztd||W5d}~XYqbXn|IdHt|dds||IdHdS)Nz#error ignored in rollback on %s: %s_pool)closedrollback ExceptionrZwarningcommitgetattrclose)r:rkrlrmZexc2r=r=r> __aexit__s" zAsyncConnection.__aexit__rr)rFrGrHcs t|f|S)z3Manipulate connection parameters before connecting.)r )r`rFrGr=r=r>rUsz&AsyncConnection._get_connection_paramscs|jr dSd|_|jdS)zClose the database connection.NT)ro_closedr5finishrhr=r=r>ruszAsyncConnection.close)binaryzAsyncCursor[Row])ryrHcCsdSr6r=)r:ryr=r=r>cursorszAsyncConnection.cursorzAsyncRowFactory[CursorRow]zAsyncCursor[CursorRow])ryr3rHcCsdSr6r=)r:ryr3r=r=r>rzs)ry scrollablewithholdz bool | NonezAsyncServerCursor[Row])nameryr{r|rHcCsdSr6r=)r:r}ryr{r|r=r=r>rzszAsyncServerCursor[CursorRow])r}ryr3r{r|rHcCsdSr6r=)r:r}ryr3r{r|r=r=r>rzs )ryr3r{r|zAsyncRowFactory[Any] | Nonez)AsyncCursor[Any] | AsyncServerCursor[Any]cCsH||s|j}|r,|j|||||d}n|j||d}|rDt|_|S)z\ Return a new `AsyncCursor` to send commands and queries to the connection. )r}r3r{r|r3)_check_connection_okr3r2r1BINARYformat)r:r}ryr3r{r|curr=r=r>rzs )prepareryrz Params | None)queryrarryrHc s^z*|}|rt|_|j|||dIdHWStjk rX}z|dW5d}~XYnXdS)z8Execute a query and return a cursor to read its results.)rN)rzrrexecuterTrXr^)r:rrarryrrfr=r=r>rs zAsyncConnection.executec s8|j4IdH||IdHW5QIdHRXdS)z/Commit any pending transaction to the database.N)r9waitZ _commit_genrhr=r=r>rsszAsyncConnection.commitc s8|j4IdH||IdHW5QIdHRXdS)z2Roll back to the start of any pending transaction.N)r9rZ _rollback_genrhr=r=r>rpszAsyncConnection.rollbackg>@rJfloat)rKrHcsF|s dStr2tj|j|dtdIdHnt|jIdHdS)aCancel the current operation on the connection. :param timeout: raise a `~errors.CancellationTimeout` if the cancellation request does not succeed within `timeout` seconds. Note that a successful cancel attempt on the client is not a guarantee that the server will successfully manage to cancel the operation. This is a non-blocking version of `~Connection.cancel()` which leverages a more secure and improved cancellation feature of the libpq, which is only available from version 17. If the underlying libpq is older than version 17, the method will fall back to using the same implementation of `!cancel()`. NrJrL) Z_should_cancelr'Zhas_cancel_saferrVZ _cancel_genrWr-cancel)r:rKr=r=r> cancel_safes zAsyncConnection.cancel_safe@c sJz|j|dIdHWn.tk rD}ztd|W5d}~XYnXdS)NrJzquery cancellation failed: %s)rrqrZrr)r:rKrfr=r=r> _try_cancel5szAsyncConnection._try_cancelz str | NonezAsyncIterator[AsyncTransaction])savepoint_nameforce_rollbackrHc Cst|||}|jr||4IdHL|4IdH.|4IdH |VW5QIdHRXW5QIdHRXW5QIdHRXn$|4IdH |VW5QIdHRXdS)a Start a context block with a new transaction or nested transaction. :param savepoint_name: Name of the savepoint used to manage a nested transaction. If `!None`, one will be chosen automatically. :param force_rollback: Roll back the transaction at the end of the block even if there were no error (e.g. to try a no-op process). :rtype: AsyncTransaction N)r%r4pipeline)r:rrZtxr=r=r> transaction;s 28zAsyncConnection.transaction)rK stop_afterz float | NonezAsyncGenerator[Notify, None])rKrrHc Cs |dk rt|}t|t}nd}t}d}|j4IdH|jj}z|jt|j|dIdH}Wn.tj k r}z| dW5d}~XYnX|D]0} t | j || j || j} | V|d7}q|dk r||krq|rBtt|t}|dkrBqqBW5QIdHRXdS)a Yield `Notify` objects as soon as they are received from the database. :param timeout: maximum amount of time to wait for notifications. `!None` means no timeout. :param stop_after: stop after receiving this number of notifications. You might actually receive more than this number if more than one notifications arrives in the same packet. NrrLr g)rminrWr9r5 _encodingrr$rTrXr^r+ZrelnamedecodeextraZbe_pid) r:rKrdeadlinerMZ nreceivedencnsrfZpgnnr=r=r>r$Ps4    zAsyncConnection.notifieszAsyncIterator[AsyncPipeline]cCs|j4IdH*||j}|dkr4t|}|_W5QIdHRXz(|4IdH |VW5QIdHRXW5|jdkr|j4IdH||jkstd|_W5QIdHRXXdS)zrs zAsyncConnection.pipelinez PQGen[RV]r)rerMrHc sztj||jj|dIdHWStk r|jjtkr|jddIdHztj||jj|dIdHWntj k r~YnXYnXdS)z Consume a generator operating on the connection. The function must be used on generators that don't change connection fd (i.e. not on connect and reset). rLNrrJ) rZ wait_asyncr5socket _INTERRUPTEDtransaction_statusACTIVErrTZ QueryCanceled)r:rerMr=r=r>rs zAsyncConnection.wait)valuerHcCs|ddS)NrA _no_set_asyncr:rr=r=r>_set_autocommits zAsyncConnection._set_autocommitc s:|j4IdH|||IdHW5QIdHRXdS)z6Method version of the `~Connection.autocommit` setter.N)r9rZ_set_autocommit_genrr=r=r>set_autocommitszAsyncConnection.set_autocommitzIsolationLevel | NonecCs|ddS)NZisolation_levelrrr=r=r>_set_isolation_levels z$AsyncConnection._set_isolation_levelc s:|j4IdH|||IdHW5QIdHRXdS)z;Method version of the `~Connection.isolation_level` setter.N)r9rZ_set_isolation_level_genrr=r=r>set_isolation_levelsz#AsyncConnection.set_isolation_levelcCs|ddS)NZ read_onlyrrr=r=r>_set_read_onlys zAsyncConnection._set_read_onlyc s:|j4IdH|||IdHW5QIdHRXdS)z5Method version of the `~Connection.read_only` setter.N)r9rZ_set_read_only_genrr=r=r> set_read_onlyszAsyncConnection.set_read_onlycCs|ddS)NZ deferrablerrr=r=r>_set_deferrables zAsyncConnection._set_deferrablec s:|j4IdH|||IdHW5QIdHRXdS)z6Method version of the `~Connection.deferrable` setter.N)r9rZ_set_deferrable_genrr=r=r>set_deferrableszAsyncConnection.set_deferrable) attributerHcCstd|d|ddS)Nz'the zD property is read-only on async connections: please use 'await .set_z ()' instead.)AttributeError)r:rr=r=r>rszAsyncConnection._no_set_asyncz Xid | str)xidrHc s:|j4IdH|||IdHW5QIdHRXdS)zO Begin a TPC transaction with the given transaction ID `!xid`. N)r9rZ_tpc_begin_genr:rr=r=r> tpc_beginszAsyncConnection.tpc_beginc srz8|j4IdH||IdHW5QIdHRXWn4tjk rl}ztt|dW5d}~XYnXdS)zV Perform the first phase of a transaction started with `tpc_begin()`. N)r9rZ_tpc_prepare_genrTZObjectNotInPrerequisiteStateZNotSupportedErrorrD)r:rfr=r=r> tpc_prepares (zAsyncConnection.tpc_preparezXid | str | Nonec s<|j4IdH||d|IdHW5QIdHRXdS)z: Commit a prepared two-phase transaction. NZCOMMITr9rZ_tpc_finish_genrr=r=r> tpc_commitszAsyncConnection.tpc_commitc s<|j4IdH||d|IdHW5QIdHRXdS)z= Roll back a prepared two-phase transaction. NZROLLBACKrrr=r=r> tpc_rollbackszAsyncConnection.tpc_rollbackz list[Xid]c s||jj}|jttjd4IdH(}|tIdH| IdH}W5QIdHRX|t kr|jjt kr| IdH|S)Nr~) Z _check_tpcinforrzrrZ _from_recordrZ_get_recover_queryZfetchallIDLEINTRANSrp)r:statusrresr=r=r> tpc_recoverszAsyncConnection.tpc_recover)r?)r?)N)NF)N)N),__name__ __module__ __qualname____doc____annotations__rrrrr8 classmethodrgrirvrUrur rzrrsrprrr rr$rrWrrrrrrrrrrrrrrr __classcell__r=r=r;r>r0>s   "=   %2 r0)Tr __future__rloggingtimertypesrtypingrrrrr r contextlibr r?r rrTrabcrrrrrrrZ_tpcrZrowsrrrrZadaptrZ_enumsr_compatrrFrr r!r"r4r# generatorsr$rr%Z cursor_asyncr&Z _capabilitiesr'Z server_cursorr(Z_connection_baser)r*r+rPrRr,r-Zpq.abcr.rWZFormatZTEXTrZTransactionStatusrrrZCancelledErrorKeyboardInterruptr getLoggerrZr0r=r=r=r>sN        $