o /i5 @sUddlZddlZddlZddlZddlZddlZddlZddlmZm Z ddl m Z ddl m Z mZmZddlmZmZmZmZmZmZmZmZmZmZddlmZmZmZddlm Z m!Z!m"Z"m#Z#m$Z$m%Z%dd l&m'Z'm(Z(m)Z)m*Z*dd l+m,Z,dd l-m.Z.dd l/m0Z0m1Z1dd l2m3Z3m4Z4ddl5m6Z6m7Z7m8Z8m9Z9m:Z:m;Z;mZ>ddl?m@Z@mAZAmBZBmCZCddlDmEZEddlFmGZGmHZHmIZImJZJmKZKmLZLmMZMmNZNmOZOeIrddlPZPddlPmQZQndZPdZQeHrddlRZRdZSdZTdZUdZVdZWeXZYeee)e*e(feZd<eHre(Z[ne)Z[GdddZ\GdddZ]GdddZ^Gdd d Z_Gd!d"d"e_e^Z`Gd#d$d$e`ZaGd%d&d&e_e^ZbGd'd(d(eaZcGd)d*d*e`Zdd+Zed,d-Zfd.d/Zgeheieiefefejehehefegegeid0 Zkd1d2Zled3d4d5ZmGd6d7d7eZnGd8d9d9ZoGd:d4d4eoenZpGd;d<d}t|}t||ksZ||ksZt|t rtt|t t|t f}||||t }qEt|t t|t |t f}qE|||SrE)rHrIrJrKrL SYM_EMPTYjoinSYM_STARlenSYM_CRLFrbmap memoryview SYM_DOLLARrM)rTrDrUbuffrcarg arg_lengthrYrYrZr[vs<"      zPythonRespSerializer.packr`N)r\r]r^rdr[rYrYrYrZr_qs  r_c@s@eZdZeddZeddZeddZeddZed d Zed d Z ed dZ eddZ eddZ ed2ddZ eddZed3ddZe d4dddddZedd Zed!d"Zeed#eeeefeeefffd$d%Zed&efd'd(Zed)d*Zed+d,Zed-d.Zed/d0Zd1S)5ConnectionInterfacecCdSrarYrTrYrYrZ repr_pieceszConnectionInterface.repr_piecescCrrrarYrTcallbackrYrYrZregister_connect_callbackruz-ConnectionInterface.register_connect_callbackcCrrrarYrvrYrYrZderegister_connect_callbackruz/ConnectionInterface.deregister_connect_callbackcCrrrarYrT parser_classrYrYrZ set_parserruzConnectionInterface.set_parsercCrrrarYrsrYrYrZ get_protocolruz ConnectionInterface.get_protocolcCrrrarYrsrYrYrZconnectruzConnectionInterface.connectcCrrrarYrsrYrYrZ on_connectruzConnectionInterface.on_connectcGrrrarYrTrDrYrYrZ disconnectruzConnectionInterface.disconnectcCrrrarYrsrYrYrZ check_healthruz ConnectionInterface.check_healthTcCrrrarYrTcommandrrYrYrZsend_packed_commandruz'ConnectionInterface.send_packed_commandcOrrrarYrTrDkwargsrYrYrZ send_commandruz ConnectionInterface.send_commandrcCrrrarYrTtimeoutrYrYrZcan_readruzConnectionInterface.can_readFdisconnect_on_error push_requestcCrrrarYrTdisable_decodingrrrYrYrZ read_responsez!ConnectionInterface.read_responsecGrrrarYrrYrYrZrOruz ConnectionInterface.pack_commandcCrrrarYrTcommandsrYrYrZ pack_commandsruz!ConnectionInterface.pack_commandsr`cCrrrarYrsrYrYrZhandshake_metadataz&ConnectionInterface.handshake_metadatatokencCrrrarYrTrrYrYrZset_re_auth_tokenruz%ConnectionInterface.set_re_auth_tokencCrrrarYrsrYrYrZre_authruzConnectionInterface.re_authcCdS)z Mark the connection to be reconnected on the next command. This is useful when a connection is moved to a different node. NrYrsrYrYrZmark_for_reconnectsz&ConnectionInterface.mark_for_reconnectcCr)zG Returns True if the connection should be reconnected. NrYrsrYrYrZshould_reconnectz$ConnectionInterface.should_reconnectcCr)z3 Reset the internal flag to False. NrYrsrYrYrZreset_should_reconnectrz*ConnectionInterface.reset_should_reconnectNTrF)r\r]r^rrtrxryr|r}r~rrrrrrrrOrpropertyrr bytesrIrr rrrrrrYrYrYrZrqs`              *   rqc@seZdZdZdejdddddfdeedeedddee dee d ee d ee d ee e effd d Zede e effddZedeejfddZede e e ffddZeede fddZejede fddZeedee e e ffddZejedee e e ffddZeedee e e ffddZejedee e e ffddZedd Ze !dFd"d!d#d$d%Zed&d'Z     dGdeed ee e effd(d)Zdefd*d+ZdHd,d-Z "dHdefd.d/Z dee fd0d1Z!edefd2d3Z"e"jdId5d3Z"d6d7Z#dJd8ee fd9d:Z$dJd;ee fdee e e'fd?ee fd@dAZ( ! !dKdBe)dCe)fdDdEZ*dS)L$MaintNotificationsAbstractConnectiona2 Abstract class for handling maintenance notifications logic. This class is expected to be used as base class together with ConnectionInterface. This class is intended to be used with multiple inheritance! All logic related to maintenance notifications is encapsulated in this class. Nmaint_notifications_config maint_notifications_pool_handlermaintenance_stater/maintenance_notification_hashorig_host_addressorig_socket_timeoutorig_socket_connect_timeoutparserc Cs(||_||_||_||||||dS)aw Initialize the maintenance notifications for the connection. Args: maint_notifications_config (MaintNotificationsConfig): The configuration for maintenance notifications. maint_notifications_pool_handler (Optional[MaintNotificationsPoolHandler]): The pool handler for maintenance notifications. maintenance_state (MaintenanceState): The current maintenance state of the connection. maintenance_notification_hash (Optional[int]): The current maintenance notification hash of the connection. orig_host_address (Optional[str]): The original host address of the connection. orig_socket_timeout (Optional[float]): The original socket timeout of the connection. orig_socket_connect_timeout (Optional[float]): The original socket connect timeout of the connection. parser (Optional[Union[_HiredisParser, _RESP3Parser]]): The parser to use for maintenance notifications. If not provided, the parser from the connection is used. This is useful when the parser is created after this object. N)rrr$_configure_maintenance_notifications) rTrrrrrrrrrYrYrZrdsz-MaintNotificationsAbstractConnection.__init__r`cCrrrarYrsrYrYrZ _get_parser<ruz0MaintNotificationsAbstractConnection._get_parsercCrrrarYrsrYrYrZ _get_socket@ruz0MaintNotificationsAbstractConnection._get_socketcCr) Returns: The RESP protocol version, or ``None`` if the protocol is not specified, in which case the server default will be used. NrYrsrYrYrZr}Dsz1MaintNotificationsAbstractConnection.get_protocolcCrrrarYrsrYrYrZhostMrz)MaintNotificationsAbstractConnection.hostrWcCrrrarYrTrWrYrYrZrRrcCrrrarYrsrYrYrZsocket_timeoutWrz3MaintNotificationsAbstractConnection.socket_timeoutcCrrrarYrrYrYrZr\rcCrrrarYrsrYrYrZsocket_connect_timeoutarz;MaintNotificationsAbstractConnection.socket_connect_timeoutcCrrrarYrrYrYrZrfrcOrrrarYrrYrYrZrkruz1MaintNotificationsAbstractConnection.send_commandFTrcCrrrarYrrYrYrZrorz2MaintNotificationsAbstractConnection.read_responsecGrrrarYrrYrYrZryruz/MaintNotificationsAbstractConnection.disconnectcCs|jr|jjsd|_d|_dS|stdt|ts#t|ts#td|r1||_|j |nd|_t ||j|_|jrE| |jj | |jj |rP|n|j|_|rX|n|j|_|rc||_dS|j|_dS)z Enable maintenance notifications by setting up handlers and storing original connection parameters. Should be used ONLY with parsers that support push notifications. NzBTo configure maintenance notifications, a parser must be provided!zLMaintenance notifications are only supported with hiredis and RESP3 parsers!)renabled!_maint_notifications_pool_handler'_maint_notifications_connection_handlerr,rHrrget_handler_for_connectionset_connectionr1set_node_moving_push_handlerhandle_notificationset_maintenance_push_handlerrrrrrr)rTrrrrrrYrYrZr}sN   zIMaintNotificationsAbstractConnection._configure_maintenance_notificationscCs`|}||||j||_|js)t||j|_| |jjdS|j|j_dSra) rrrrrrrr1configr)rTr%maint_notifications_pool_handler_copyrYrYrZ3set_maint_notifications_pool_handler_for_connections"  zXMaintNotificationsAbstractConnection.set_maint_notifications_pool_handler_for_connectioncCsR|dvr|jr!|jjr#|jr%t|dr'|j|j|ddSdSdSdSdSdS)NrA2r)rr)r}rrrhasattr!_enable_maintenance_notifications)rTrrYrYrZ0activate_maint_notifications_handling_if_enableds  zUMaintNotificationsAbstractConnection.activate_maint_notifications_handling_if_enabledc Csz3t|dd}|durtd|||}|jdddd|j|d|}|r-t|dkr1td WdStyc}z$t |trW|j d krWd dl }| t }|d |nWYd}~dSd}~ww) NrzaCannot enable maintenance notifications for connection object that doesn't have a host attribute.CLIENTMAINT_NOTIFICATIONSONzmoving-endpoint-typerOKz4The server doesn't support maintenance notificationsautorz,Failed to enable maintenance notifications: )getattr ValueErrorget_endpoint_typerrWrr<r- ExceptionrHrlogging getLoggerr\warning) rTrrr endpoint_typeresponseerloggerrYrYrZrs@    zFMaintNotificationsAbstractConnection._enable_maintenance_notificationsc Csz|}|dur|}|rt|dkr|dWSWn ttfy&Ynwz+t|dd}t|dd}|rMt||tjtj }|rPt |dddWSWdSWdStttj fy`YdSw) a Extract the resolved IP address from an established connection or resolve it from the host. First tries to get the actual IP from the socket (most accurate), then falls back to DNS resolution if needed. Args: connection: The connection object to extract the IP from Returns: str: The resolved IP address, or None if it cannot be determined Nrrr localhostport) r getpeernamerhAttributeErrorOSErrorrsocket getaddrinfo AF_UNSPEC SOCK_STREAMrIgaierror)rT conn_socket peer_addrrr addr_inforYrYrZget_resolved_ip"s6     z4MaintNotificationsAbstractConnection.get_resolved_ipcC|jSra_maintenance_statersrYrYrZrUz6MaintNotificationsAbstractConnection.maintenance_statestatecC ||_dSrar)rTrrYrYrZrY cCs|}|r |dSdS)z: Returns the peer name of the connection. rN)rr)rTrrYrYrZr]s z0MaintNotificationsAbstractConnection.getpeernamerelaxed_timeoutcCs:|}|r|dkr |n|j}||||dSdS)N)rr settimeoutupdate_parser_timeout)rTrrrrYrYrZupdate_current_socket_timeoutfs  zBMaintNotificationsAbstractConnection.update_current_socket_timeoutrcCsL|}|r |jr"t|tr|r||j_dSt|tr$||_dSdSdSdSra)r_bufferrHrrr_socket_timeout)rTrrrYrYrZrms    z:MaintNotificationsAbstractConnection.update_parser_timeouttmp_host_addresstmp_relaxed_timeoutcCs2|r |tkr t||_|dkr||_||_dSdS)zd The value of SENTINEL is used to indicate that the property should not be updated. rN)SENTINELrIrrr)rTrrrYrYrZset_tmp_settingsus   z5MaintNotificationsAbstractConnection.set_tmp_settingsreset_host_addressreset_relaxed_timeoutcCs(|r|j|_|r|j|_|j|_dSdSra)rrrrrr)rTrrrYrYrZreset_tmp_settingss  z7MaintNotificationsAbstractConnection.reset_tmp_settingsr)NNNNNr)rr/raFF)+r\r]r^__doc__r/NONErr0r2intrIfloatrrrrdrrrrr}rrsetterrrrrrrrrrrrrrrrobjectrboolrrYrYrYrZr s   '   K ! &3    rc:@sneZdZdZdddddedddeddddeddddd ddddejddddfd e d e e d e e d e e de deeeeefde de de de de de e de e de e de e deedfde egdfde ede e de egdfde ede ed e ed!d"d#e e d$e e d%e e d&e e f8d'd(Zd)d*Zed+d,Zd-d.Zd/d0Zd1d2Zd3d4Z d5d6Z!d7ee"e#e$ffd8d9Z%d:d;Z& e fd?d@Z'edAdBZ(edCdDZ)dEdFZ*dGdHZ+dzd=e fdIdJZ,dKdLZ-dMdNZ.dOdPZ/dQdRZ0dSdTZ1dUdVZ2dWdXZ3dzdYdZZ4d[d\Z5d{d]d^Z6 d|de>j?djeedle@fdmdnZAdodpZBd7e eCjCfdqdrZDe;d7e ee e ffdsdtZEeEj?dje ee e ffdudtZEe;d7e ee e ffdvdwZFeFj?dje ee e ffdxdwZFdS)}AbstractConnectionz0Manages communication to and from a Redis serverrNFutf-8strictizredis-pyrAdbpasswordrrretry_on_timeoutretry_on_errorencodingencoding_errorsdecode_responsessocket_read_sizehealth_check_interval client_namelib_name lib_versionusernameretryredis_connect_funccredential_providerprotocolcommand_packerevent_dispatcherrrrr/rrrrc Cs|s|r |dur td|durt|_n||_t|_||_| |_||_||_ ||_ ||_ ||_ ||_ |dur;|}||_||_|turHg}nt|}|rS|t||_|s[|jrx|durgttd|_nt||_|jrw|j|jnttd|_| |_d|_||_t||| |_d|_ d|_!| |_"g|_#d|_$d|_%z-zt&|}Wnt'yt(}Yn t)yt*dwW|dks|dkrt*d ||_+n|dks|dkrt*d ||_+w|j+dkr| t,krt-} |.| |/||_0d |_1t23|||||||||j4 dS) a2 Initialize a new Connection. To specify a retry policy for specific errors, first set `retry_on_error` to a list of the error/s to retry on, then set `retry` to a valid `Retry` object. To retry on TimeoutError, `retry_on_timeout` can also be set to `True`. Nz'username' and 'password' cannot be passed along with 'credential_provider'. Please provide only one of the following arguments: 1. 'password' and (optional) 'username' 2. 'credential_provider'rripzprotocol must be an integerrAzprotocol must be either 2 or 3F)5r*r%_event_dispatcherosgetpidpidrrrrrr rr_socket_connect_timeoutr rlistrMr.r r3r!rcopydeepcopyupdate_supported_errorsrnext_health_checkrrencoderr_sock_socket_read_size_connect_callbacksrb_re_auth_tokenrrPDEFAULT_RESP_VERSIONrr)rrrr|_construct_command_packer_command_packer_should_reconnectrrd_parser) rTrr rrr r r r rr{rrrrrrrrrrrrrrrrrrrretry_on_errors_listprYrYrZrds)          zAbstractConnection.__init__cCs8ddd|D}d|jjd|jjd|dS)N,cSg|] \}}|d|qS=rY.0kvrYrYrZ z/AbstractConnection.__repr__..<.(z)>)rfrt __class__r]r\)rT repr_argsrYrYrZ__repr__s zAbstractConnection.__repr__cCrrrarYrsrYrYrZrtruzAbstractConnection.repr_piecescCs$z|WdStyYdSwra)rrrsrYrYrZ__del__s  zAbstractConnection.__del__cCs&|dur|Str tSt|j|jjSra)r5rCr_rbr&rK)rTpackerrYrYrZr,"s z,AbstractConnection._construct_command_packercCs(t|}||jvr|j|dSdS)a^ Register a callback to be called when the connection is established either initially or reconnected. This allows listeners to issue commands that are ephemeral to the connection, for example pub/sub subscription or key tracking. The callback must be a _method_ and will be kept as a weak reference. N)weakref WeakMethodr)rM)rTrwwmrYrYrZrx*s  z,AbstractConnection.register_connect_callbackcCs.z |jt|WdStyYdSw)z De-register a previously registered callback. It will no-longer receive notifications on connection events. Calling this is not required when the listener goes away, since the callbacks are kept as weak methods. N)r)removerDrErrvrYrYrZry6s  z.AbstractConnection.deregister_connect_callbackcCs||jd|_dS)z Creates a new instance of parser_class with socket size: _socket_read_size and assigns it to the parser for the connection :param parser_class: The required parser class )rN)r(r/rzrYrYrZr|AszAbstractConnection.set_parserr`cCrra)r/rsrYrYrZrIzAbstractConnection._get_parsercCs|jdddS)z5Connects to the Redis server if not already connectedTrN)connect_check_healthrsrYrYrZr~LszAbstractConnection.connectTrretry_socket_connectc sjrdSz|rjfddfdd}n}Wntjy)tdty;}zt |d}~ww|_zj durLj |dn Wn t y^ wddjD_jD] }|}|rv|qkdS)NcSra)_connectrYrsrYrZXz9AbstractConnection.connect_check_health..c |SrarerrorrsrYrZrMX zTimeout connecting to serverrcSsg|]}|r|qSrYrY)r7refrYrYrZr:qsz;AbstractConnection.connect_check_health..)r'rcall_with_retryrLrrr.rr)_error_messageron_connect_check_healthr,rr))rTrrJsockrrTrwrYrsrZrIPs@    z'AbstractConnection.connect_check_healthcCrrrarYrsrYrYrZrLwruzAbstractConnection._connectcCrrrarYrsrYrYrZ _host_error{ruzAbstractConnection._host_errorcCst||Sra)r:rY)rT exceptionrYrYrZrVz!AbstractConnection._error_messagecCs|jdddS)NTr)rWrsrYrYrZrzAbstractConnection.on_connectcCs|j||j}d}|js|js|jr"|jpt|j|j}|}|r`|jdvr`t|jt r?| t |j |j_ |j|t |dkrKd|dg}|jd|jdg|Rdd i||_nv|r|jdg|Rdd iz|}Wnty|jd|d d d |}Ynwt|d krtd n?|jdvrt|jt r| t |j |j_ |j||jd|j|d ||_|jd|jkr|jd|jkrtd|j|d |jr|jdd|j|d t|d krtdz|jr |jddd|j|d |Wn tyYnwz|jr)|jddd|j|d |Wn ty4Ynw|jrO|jd|j|d t|d krQtddSdS)z=Initialize the connection, authenticate and select a databaseNrrdefaultrHELLOAUTHrFrrrzInvalid Username or PasswordsprotoprotozInvalid RESP versionrSETNAMEzError setting client nameSETINFOzLIB-NAMEzLIB-VERSELECTzInvalid Database)r/rrrr r#get_credentialsrrHrr|rEXCEPTION_CLASSESrhrrrr'r<r&getr)rrrr-rr)rTrr auth_args cred_provider auth_responserYrYrZrWs                    z*AbstractConnection.on_connect_check_healthc Gs|j|j}d|_||durdSt|jkr0z|tj Wn t t fy/Ynwz| WdSt yAYdSw)z!Disconnects from the Redis serverN) r/ on_disconnectr'rrrrshutdownr SHUT_RDWRrrPclose)rTrD conn_sockrYrYrZrs"  zAbstractConnection.disconnectcC d|_dS)NTr.rsrYrYrZr z%AbstractConnection.mark_for_reconnectcCrrarprsrYrYrZrrHz#AbstractConnection.should_reconnectcCro)NFrprsrYrYrZrrqz)AbstractConnection.reset_should_reconnectcCs*|jdddt|dkrtddS)z Send PING, expect PONG in returnPINGFrPONGz#Bad response from PING health checkN)rr<rr)rsrYrYrZ _send_pingszAbstractConnection._send_pingcC |dS)z Function to call when PING failsNrPrTrRrYrYrZ _ping_failed zAbstractConnection._ping_failedcCs2|jrt|jkr|j|j|jdSdSdS)z3Check the health of the connection with a PING/PONGN)rtime monotonicr%rrUrtrwrsrYrYrZr"szAbstractConnection.check_healthc Cs|js |jdd|r|zt|tr|g}|D]}|j|qWd Stjy5|t dt yi}z)|t |j dkrPd|j d}}n |j d}|j d}t d|d|d d }~wtyt|w) z2Send an already packed command to the Redis serverFrzTimeout writing to socketrUNKNOWNrzError z while writing to socket. r=N)r'rIrrHrIsendallrrrr.rrhrDr) BaseException)rTrritemrerrnoerrmsgrYrYrZr's2     z&AbstractConnection.send_packed_commandcOs"|j|jj||ddddS)z+Pack and send a command to the Redis serverrTrN)rr-r[rfrrYrYrZrFs   zAbstractConnection.send_commandc Cs`|j}|s ||}z|j|WSty/}z|td|d|jd}~ww)z8Poll the socket to see if there's data that can be read.Error while reading from z: N) r'r~rYr/rrrr)rD)rTrrX host_errorrrYrYrZrMszAbstractConnection.can_readrc Cs|}z|jdvr|jj||d}n|jj|d}Wn>tjy0|r)|td|tyL}z|r=|t d|d|j d}~wt yY|rX|w|j ret |j |_t|troz|~w|S)z0Read the response from a previously sent command)3r)rr)rzTimeout reading from rz : N)rYrr/rrrrr.rr)rDr}rryrzr%rHr-)rTrrrrrrrYrYrZr\s:    z AbstractConnection.read_responsecG |jj|S)rF)r-r[rrYrYrZrOrxzAbstractConnection.pack_commandc Csg}g}d}|j}|D]B}|jj|D]9}t|}||ks&||ks&t|tr4|r0|t|d}g}||ks=t|trC||q||||7}qq |rX|t||S)z.Pack multiple commands into the Redis protocolr) rbr-r[rhrHrkrMrerf) rTrrUpieces buffer_lengthrccmdchunkchunklenrYrYrZrs.   z AbstractConnection.pack_commandscCrra)rrsrYrYrZr}rHzAbstractConnection.get_protocolcCrra_handshake_metadatarsrYrYrZrrz%AbstractConnection.handshake_metadatarWcCrrarrrYrYrZrrrcCrra)r*rrYrYrZrrqz$AbstractConnection.set_re_auth_tokencCs<|jdur|d|jd|j|d|_dSdSNr_oid)r*rtry_get get_valuerrsrYrYrZrs   zAbstractConnection.re_authcCrra)r'rsrYrYrZrrHzAbstractConnection._get_socketcCrrarrsrYrYrZrrz!AbstractConnection.socket_timeoutcCrrarrrYrYrZrrcCrrar rsrYrYrZrrz)AbstractConnection.socket_connect_timeoutcCrrarrrYrYrZrr)TTrrr)Gr\r]r^rrrBr;r/rrrrIrrrr rrrrr r"r%r0r2rdrArrtrBr,rxryr|rrrrr~rIrLrYrVrrWrrrrrtrwrrrrrrOrr}rr rrrr rrrrrrrYrYrYrZrs&          '  t   ,((  rcsjeZdZdZ     dfdd Zd d Zd d Zd dZede fddZ e j de fddZ Z S) Connectionz4Manages TCP communication to and from a Redis serverrrFNrc s<||_t||_||_|pi|_||_tjdi|dSNrY)_hostrrsocket_keepalivesocket_keepalive_options socket_typesuperrd)rTrrrrrrr?rYrZrds   zConnection.__init__cCs6d|jfd|jfd|jfg}|jr|d|jf|S)Nrrrr)rrrrrMrTrrYrYrZrtszConnection.repr_piecesc Cs,d}t|j|j|jtjD]}}|\}}}}}d}zDt|||}|tjtjd|j rH|tj tj d|j D] \} } |tj| | q;||j||||j|WSty} z"| }|durz|tjWn ty|Ynw|WYd} ~ qd} ~ ww|dur|td)zCreate a TCP socket connectionNrz)socket.getaddrinfo returned an empty list)rrrrrr setsockopt IPPROTO_TCP TCP_NODELAYr SOL_SOCKET SO_KEEPALIVEritemsrrr~rrrkrlrm) rTerrresfamilysocktyper` canonnamesocket_addressrXr8r9rVrYrYrZrLs>      zConnection._connectcCs|jd|jS)N:)rrrsrYrYrZrYszConnection._host_errorr`cCrrarrsrYrYrZr rzConnection.hostrWcCrrarrrYrYrZr$r)rrFNr) r\r]r^rrdrtrLrYrrIrr __classcell__rYrYrrZrs.rc@seZdZdZdZdZdededej fddZ d d Z d d Z d dZ ddZddZddZddZddZddZddZdhddZd d!Zdid#d$Z %djdd%d&d'd(Zd)d*Zd+d,Zed-eeeefeeefffd.d/Z d0e!fd1d2Z"d3d4Z#d5d6Z$d7d8Z%d9d:Z&ed-efd;d<Z'e'j(d=efd>d<Z'ed-e)ee*e+ffd?d@Z,e,j(d=e)ee*e+ffdAd@Z,ed-e)ee*e+ffdBdCZ-e-j(d=e)ee*e+ffdDdCZ-d-e)e.j.fdEdFZ/d-e0fdGdHZ1ed-e2fdIdJZ3e3j(dKe2fdLdJZ3dMdNZ4dOdPZ5dkdRe)e*fdSdTZ6 Q QdldUe)edVe)e*fdWdXZ7 % %dmdYe8dZe8fd[d\Z9d]d^Z:d_d`Z;ded-dQfdadbZeee)e>effdfdgZ?dQS)nCacheProxyConnectionsfooz7.4.0redisconncache pool_lockc Cst|_||_|jj|_|jj|_|jj|_|j|_||_||_ t |_ d|_ d|_||jt|jtrXt||jj|jj|jj|jj|jj|jj|jj|j dSdSra)rrr_connrrrr _pool_lock_cache threadingRLock _cache_lock_current_command_cache_key_current_optionsrx_enable_tracking_callbackrHrrdrrrrrrr)rTrrrrYrYrZrd.s2       zCacheProxyConnection.__init__cC |jSra)rrtrsrYrYrZrtNrqz CacheProxyConnection.repr_piecescC|j|dSra)rrxrvrYrYrZrxQr\z.CacheProxyConnection.register_connect_callbackcCrra)rryrvrYrYrZryTr\z0CacheProxyConnection.deregister_connect_callbackcCrra)rr|rzrYrYrZr|Wr\zCacheProxyConnection.set_parsercCs t|jtr|j|dSdSra)rHrrrrTrrYrYrZrZs zHCacheProxyConnection.set_maint_notifications_pool_handler_for_connectioncCrra)rr}rsrYrYrZr}brqz!CacheProxyConnection.get_protocolcCs|j|jjdd}|dur|jjdd}|jjdd}|dur-|jjdd}|dus5|dur9tdt|}t|}||jksNt||jdkrRtddS)Nsserverserversversionversionz0Cannot retrieve information about server versionrziTo maximize compatibility with all Redis products, client-side caching is supported by Redis 7.4 or later) rr~rrfr)r9DEFAULT_SERVER_NAMEr7MIN_ALLOWED_VERSION)rT server_name server_verrYrYrZr~es"  zCacheProxyConnection.connectcC|jdSra)rrrsrYrYrZr|r[zCacheProxyConnection.on_connectcGs@|j |jWdn1swY|jj|dSra)rrflushrrrrYrYrZrs zCacheProxyConnection.disconnectcCrra)rrrsrYrYrZrr[z!CacheProxyConnection.check_healthTcCrra)rrrrYrYrZrsz(CacheProxyConnection.send_packed_commandc Os||j)|jt|ddds)d|_|jj|i| WddSWdn1s3wY|ddurCt dt|dt |dd|_|jY|j|jr|j|j}|j |jkr|j |j r~|j jdd|j srWdn1swY WddS|jt|j|jtj|jdWdn1swY|jj|i|dS) NrrY)r redis_keyskeyszCannot create cache key.Tr) cache_key cache_valuestatusconnection_ref)_process_pending_invalidationsrr is_cachablerrrrrfrrJrrrrsetrDUMMY_CACHE_VALUEr IN_PROGRESS)rTrDrentryrYrYrZrsH   z!CacheProxyConnection.send_commandrcC |j|Sra)rrrrYrYrZr zCacheProxyConnection.can_readFrcCs@|j8|jdur4|j|jdur4|j|jjtjkr4t|j|jj }d|_|WdSWdn1s>wY|j j |||d}|jG|jdur^|WdS|durs|j |jg|WdS|j|j}|durtj |_||_ |j|d|_Wd|S1swY|S)N)rrr)rrrrfrrrr"r#rrrdelete_by_cache_keysVALIDr)rTrrrrr cache_entryrYrYrZrsJ     z"CacheProxyConnection.read_responsecGrra)rrOrrYrYrZrOrz!CacheProxyConnection.pack_commandcCrra)rrrrYrYrZrrz"CacheProxyConnection.pack_commandsr`cC|jjSra)rrrsrYrYrZrz'CacheProxyConnection.handshake_metadatarcCrra)rrrrYrYrZrr\z&CacheProxyConnection.set_re_auth_tokencCrra)rrrsrYrYrZrr[zCacheProxyConnection.re_authcCrra)rrrsrYrYrZrr[z'CacheProxyConnection.mark_for_reconnectcCrra)rrrsrYrYrZrrqz%CacheProxyConnection.should_reconnectcCrra)rrrsrYrYrZrr[z+CacheProxyConnection.reset_should_reconnectcCrrarrrsrYrYrZrrzCacheProxyConnection.hostrWcC ||j_dSrarrrYrYrZr rxcCrrarrrsrYrYrZr rz#CacheProxyConnection.socket_timeoutcCrrarrrYrYrZrrxcCrrarrrsrYrYrZrrz+CacheProxyConnection.socket_connect_timeoutcCrrarrrYrYrZrrxcCst|jtr |jStd)NCMaintenance notifications are not supported by this connection type)rHrrrNotImplementedErrorrsrYrYrZrs  z CacheProxyConnection._get_socketcCst|ts td|S)a| Validate that connection instance supports maintenance notifications. With this helper method we ensure that we are working with the correct connection type. After twe validate that connection instance supports maintenance notifications we can safely return the connection instance as MaintNotificationsAbstractConnection. r)rHrrrT connectionrYrYrZ,_get_maint_notifications_connection_instance%s zACacheProxyConnection._get_maint_notifications_connection_instancecCs||j}|jSrarrrrTconrYrYrZr7s z&CacheProxyConnection.maintenance_statercCs||j}||_dSrar)rTrrrYrYrZr<s  cC||j}|Sra)rrrrrYrYrZrA z CacheProxyConnection.getpeernamecCrra)rrrrrYrYrZrErz$CacheProxyConnection.get_resolved_ipNrcCs||j}||dSra)rrr)rTrrrYrYrZrIs z2CacheProxyConnection.update_current_socket_timeoutrrcC||j}|||dSra)rrr)rTrrrrYrYrZrM z%CacheProxyConnection.set_tmp_settingsrrcCrra)rrr)rTrrrrYrYrZrUrz'CacheProxyConnection.reset_tmp_settingscCrra)rrLrsrYrYrZrL]r[zCacheProxyConnection._connectcCrra)rrYrsrYrYrZrY`r[z CacheProxyConnection._host_errorcCs(|ddd||j|jdS)NrTRACKINGr)rrr/set_invalidation_push_handler_on_invalidation_callback)rTrrYrYrZrcsz.CacheProxyConnection._enable_tracking_callbackcCs&|r|jjdd|sdSdS)NTr)rrrrsrYrYrZrhsz3CacheProxyConnection._process_pending_invalidationsdatacCsd|j%|ddur|jn|j|dWddSWddS1s+wYdS)Nr)rrrdelete_by_redis_keys)rTrrYrYrZrls  "z.CacheProxyConnection._on_invalidation_callbackrrrra)NNr)@r\r]r^rrrrqrrrrdrtrxryr|rr}r~rrrrrrrrOrrrr rrIrr rrrrrrrrrrrrrrrrr/rrrrrrrrLrYrrr rrYrYrYrZr)s    0 -(    &rcspeZdZdZ                ddeeddeedffd d Zfd d Zd dZZ S) SSLConnectionzManages SSL connections to and from the Redis server(s). This class extends the Connection class, adding SSL functionality, and making use of ssl.SSLContext (https://docs.python.org/3/library/ssl.html#ssl.SSLContext) NrequiredTFssl_include_verify_flagsr=ssl_exclude_verify_flagsc ststd||_||_|durtj}nt|tr1tjtjtj d}||vr-td|||}||_ ||_ ||_ ||_ ||_| |_|j tjkrK|nd|_| |_| |_| |_| |_||_||_||_tjdi|dS)a!Constructor Args: ssl_keyfile: Path to an ssl private key. Defaults to None. ssl_certfile: Path to an ssl certificate. Defaults to None. ssl_cert_reqs: The string value for the SSLContext.verify_mode (none, optional, required), or an ssl.VerifyMode. Defaults to "required". ssl_include_verify_flags: A list of flags to be included in the SSLContext.verify_flags. Defaults to None. ssl_exclude_verify_flags: A list of flags to be excluded from the SSLContext.verify_flags. Defaults to None. ssl_ca_certs: The path to a file of concatenated CA certificates in PEM format. Defaults to None. ssl_ca_data: Either an ASCII string of one or more PEM-encoded certificates or a bytes-like object of DER-encoded certificates. ssl_check_hostname: If set, match the hostname during the SSL handshake. Defaults to True. ssl_ca_path: The path to a directory containing several CA certificates in PEM format. Defaults to None. ssl_password: Password for unlocking an encrypted private key. Defaults to None. ssl_validate_ocsp: If set, perform a full ocsp validation (i.e not a stapled verification) ssl_validate_ocsp_stapled: If set, perform a validation on a stapled ocsp response ssl_ocsp_context: A fully initialized OpenSSL.SSL.Context object to be used in verifying the ssl_ocsp_expected_cert ssl_ocsp_expected_cert: A PEM armoured string containing the expected certificate to be returned from the ocsp verification service. ssl_min_version: The lowest supported SSL version. It affects the supported SSL versions of the SSLContext. None leaves the default provided by ssl module. ssl_ciphers: A string listing the ciphers that are allowed to be used. Defaults to None, which means that the default ciphers are used. See https://docs.python.org/3/library/ssl.html#ssl.SSLContext.set_ciphers for more information. Raises: RedisError z$Python wasn't built with SSL supportN)noneoptionalrz+Invalid SSL Certificate Requirements Flag: FrY)r6r,keyfilecertfilessl CERT_NONErHrI CERT_OPTIONAL CERT_REQUIRED cert_reqsrrca_certsca_dataca_pathcheck_hostnamecertificate_passwordssl_validate_ocspssl_validate_ocsp_stapledssl_ocsp_contextssl_ocsp_expected_certssl_min_version ssl_ciphersrrd)rT ssl_keyfile ssl_certfile ssl_cert_reqsrr ssl_ca_certs ssl_ca_datassl_check_hostname ssl_ca_path ssl_passwordrrrrrrr CERT_REQSrrYrZrd{s@- zSSLConnection.__init__c s4t}z||WSttfy|w)zN Wrap the socket with SSL support, handling potential errors. )rrL_wrap_socket_with_sslrr,rmrTrXrrYrZrLs  zSSLConnection._connectc Cst}|j|_|j|_|jr|jD] }|j|O_q|jr-|jD] }|j|M_q"|js3|j r>|j |j|j |j d|j dusM|j dusM|jdurX|j|j |j |jd|jdura|j|_|jrj||j|jdurwtdurwtd|jr|jrtd|j||jd}|jrd dl}d d lm}|jdur|j|jj }|!|j|"|j n|j}|#||j$|j%|t&&}|'|(|j|j)f|*|+|S|jdurtrd d lm,} | ||j|j)|j } | -r|St.d |S)z Wraps the socket with SSL support. Args: sock: The plain socket to wrap with SSL. Returns: An SSL wrapped socket. )rrr N)cafilecapathcadataTFzcryptography is not installed.zKEither an OCSP staple or pure OCSP connection must be validated - not both.)server_hostnamerr)ocsp_staple_verifier) OCSPVerifierzocsp validation error)/rcreate_default_contextrr verify_moder verify_flagsrrrload_cert_chainrrrrload_verify_locationsrminimum_versionr set_ciphersrr4r,r wrap_socketrOpenSSLocsprrSSLContext SSLv23_METHODuse_certificate_fileuse_privatekey_fileset_ocsp_client_callbackrrr request_ocspr~r do_handshakerkris_validr)) rTrXcontextflagsslsockrr staple_ctxrrorYrYrZr sp               z#SSLConnection._wrap_socket_with_ssl)NNrNNNNTNNFFNNNN) r\r]r^rrr rdrLr rrYrYrrZrus2   Q rcs:eZdZdZd fdd ZddZdd Zd d ZZS) UnixDomainSocketConnectionz4Manages UDS communication to and from a Redis serverNc s"tjdi|||_||_dSr)rrdpathr)rTr/rrrrYrZrd0s z#UnixDomainSocketConnection.__init__cCs.d|jfd|jfg}|jr|d|jf|S)Nr/rr)r/rrrMrrYrYrZrt5sz&UnixDomainSocketConnection.repr_piecesc Cszttjtj}||jz||jWnty4z|tj Wn ty.Ynw| w||j |S)z&Create a Unix domain socket connection) rAF_UNIXrrrr~r/rrkrlrmrrrYrYrZrL;s    z#UnixDomainSocketConnection._connectcCrra)r/rsrYrYrZrYLrHz&UnixDomainSocketConnection._host_error)r.N) r\r]r^rrdrtrLrYrrYrYrrZr--s r-)0FFALSENNOcCs6|dus|dkr dSt|tr|tvrdSt|S)Nr.F)rHrIupper FALSE_STRINGSr)rWrYrYrZto_boolSs r8cCs\|dddd}g}|dD]}|}tt|s#td||tt|q|S)N[r.]r2zInvalid ssl verify flag: )replacerLstriprr=rrMr)rWverify_flags_strrr)rYrYrZparse_ssl_verify_flags[s r>) rrrrr r max_connectionsrr rrrc Cs|ds|ds|dstdt|}i}t|jD]7\}}|rWt|dkrWt|d}t |}|rSz||||<Wq t tfyRtd|dw|||<q |j rbt|j |d<|j rlt|j |d <|j d kr|jr{t|j|d <t|d <|S|jrt|j|d <|jrt|j|d<|jrd|vrztt|jdd|d<Wn ttfyYnw|j dkrt|d <|S)Nzredis://z rediss://zunix://zRRedis URL must specify one of the following schemes (redis://, rediss://, unix://)rzInvalid value for 'z' in connection URL.rr unixr/connection_classrrr/r.rediss) startswithrrrqueryrrhrURL_QUERY_ARGUMENT_PARSERSrfrPrr schemer/r-hostnamerrr;rr)urlrnamerWrrYrYrZ parse_urlysZ    rK_CPConnectionPool)boundc@seZdZeddZeddZeedgdddd eed e fd d Z ed dZ ede fddZ edde fddZeddZedefddZedefddZdS) ConnectionPoolInterfacecCrrrarYrsrYrYrZr}ruz$ConnectionPoolInterface.get_protocolcCrrrarYrsrYrYrZresetruzConnectionPoolInterface.reset*)Use get_connection() without args instead5.3.0 args_to_warnreasonr command_namer`cOrrrarY)rTrWroptionsrYrYrZget_connections z&ConnectionPoolInterface.get_connectioncCrrrarYrsrYrYrZ get_encoderruz#ConnectionPoolInterface.get_encoderrcCrrrarYrrYrYrZreleaseruzConnectionPoolInterface.releaseTinuse_connectionscCrrrarY)rTr\rYrYrZrruz"ConnectionPoolInterface.disconnectcCrrrarYrsrYrYrZrmruzConnectionPoolInterface.closercCrrrarY)rTrrYrYrZ set_retryruz!ConnectionPoolInterface.set_retryrcCrrrarYrrYrYrZre_auth_callbackruz(ConnectionPoolInterface.re_auth_callbackNr)r\r]r^rr}rPr8rrIrqrYrZr[rrrmr3r]r r^rYrYrYrZrOs8    rOc@seZdZdZ d9deefddZeede e e ffddZ e j ed e e e ffd dZ edejfd d Zeded fddZeded fddZddZdefddZdefddZdefddZ   d:dd deddee d eedef d!d"Z     # # #d;dd d$ed%d&eed'ee d(eed)ed*ed+efd,d-Z        # # # .dMaintNotificationsAbstractConnectionPool._get_free_connectionscCrrrarYrsrYrYrZ_get_in_use_connectionssz@MaintNotificationsAbstractConnectionPool._get_in_use_connectionscCs|jr|jjnd}|o |jS)a Returns: True if the maintenance notifications are enabled, False otherwise. The maintenance notifications config is stored in the pool handler. If the pool handler is not set, the maintenance notifications are not enabled. N)rrrrTrrYrYrZmaint_notifications_enableds   zDMaintNotificationsAbstractConnectionPool.maint_notifications_enabledcCsN|r |js td|jst|||_n||j_||j||jdS)a# Updates the maintenance notifications configuration. This method should be called only if the pool was created without enabling the maintenance notifications and in a later point in time maintenance notifications are requested to be enabled. z Update the settings for a single connection. N)r)r)rr)rrrrr) rTrrrrsrrtrrrYrYrZupdate_connection_settingss  zCMaintNotificationsAbstractConnectionPool.update_connection_settingsTinclude_free_connectionsc Cs|K|D]} || |||r |j| |||||| | dq | r?|D]} || |||r>|j| |||||| | dq'WddSWddS1sRwYdS)a Update the settings for all matching connections in the pool. This method does not create new connections. This method does not affect the connection kwargs. :param state: The maintenance state to set for the connection. :param maintenance_notification_hash: The hash of the maintenance notification to set for the connection. :param host_address: The host address to set for the connection. :param relaxed_timeout: The relaxed timeout to set for the connection. :param matching_address: The address to match for the connection. :param matching_notification_hash: The notification hash to match for the connection. :param matching_pattern: The pattern to match for the connection. :param update_notification_hash: Whether to update the notification hash for the connection. :param reset_host_address: Whether to reset the host address to the original address. :param reset_relaxed_timeout: Whether to reset the relaxed timeout to the original timeout. :param include_free_connections: Whether to include free/available connections. )rrrsrrtrrN)rdrfrrrure) rTrrrsrrprqrmrtrrrvrrYrYrZupdate_connections_settingssT #  "zDMaintNotificationsAbstractConnectionPool.update_connections_settingscKs|j|dS)z Update the connection kwargs for all future connections. This method updates the connection kwargs for all future connections created by the pool. Existing connections are not affected. N)rcrkrTrrYrYrZupdate_connection_kwargs s zAMaintNotificationsAbstractConnectionPool.update_connection_kwargsmoving_address_srccCR||D] }||d|r|q WddS1s"wYdS)z Mark all active connections for reconnect. This is used when a cluster node is migrated to a different address. :param moving_address_src: The address of the node that is being moved. rlN)rdrfrrrrTrzrrYrYrZ'update_active_connections_for_reconnect  "zPMaintNotificationsAbstractConnectionPool.update_active_connections_for_reconnectcCr{)z Disconnect all free/available connections. This is used when a cluster node is migrated to a different address. :param moving_address_src: The address of the node that is being moved. rlN)rdrerrrr|rYrYrZdisconnect_free_connections( r~zDMaintNotificationsAbstractConnectionPool.disconnect_free_connectionsra)rlNN)NNNNFFF) NNNNNNrlFFFT)#r\r]r^rrr0rdrrr rIrrcrrrrdr rerfrhrjr2rarir rrrrrrurwryr}rrYrYrYrZr_s    !     %    I r_c@sneZdZdZedeededefddZe dddfde e d e e d e e fd d Zdefd dZedeeeffddZejdeeeffddZddZd?ddZd?ddZedgdddd@dAddZdefd d!ZdBd#d$ZdCd&d'Zd%dde fd(d)ZdDd+eddfd,d-Z d?d.d/Z!d0e"ddfd1d2Z#d3e$fd4d5Z%d6d7Z&d8d9Z'd:d;Z(dZ*dS)ErMa> Create a connection pool. ``If max_connections`` is set, then this object raises :py:class:`~redis.exceptions.ConnectionError` when the pool's limit is reached. By default, TCP connections are created unless ``connection_class`` is specified. Use class:`.UnixDomainSocketConnection` for unix sockets. :py:class:`~redis.SSLConnection` can be used for SSL enabled connections. If ``maint_notifications_config`` is provided, the connection pool will support maintenance notifications. Maintenance notifications are supported only with RESP3. If the ``maint_notifications_config`` is not provided but the ``protocol`` is 3, the maintenance notifications will be enabled by default. Any additional keyword arguments are passed to the constructor of ``connection_class``. clsrIr`cKs4t|}d|vr|d|d<|||di|S)a Return a connection pool configured from the given URL. For example:: redis://[[username]:[password]]@localhost:6379/0 rediss://[[username]:[password]]@localhost:6379/0 unix://[username@]/path/to/socket.sock?db=0[&password=password] Three URL schemes are supported: - `redis://` creates a TCP socket connection. See more at: - `rediss://` creates a SSL wrapped TCP socket connection. See more at: - ``unix://``: creates a Unix Domain Socket connection. The username, password, hostname, path and all querystring values are passed through urllib.parse.unquote in order to replace any percent-encoded values with their corresponding characters. There are several ways to specify a database number. The first value found will be used: 1. A ``db`` querystring option, e.g. redis://localhost?db=0 2. If using the redis:// or rediss:// schemes, the path argument of the url, e.g. redis://localhost/0 3. A ``db`` keyword argument to this function. If none of these options are specified, the default db=0 is used. All querystring options are cast to their appropriate Python types. Boolean arguments can be specified with string values "True"/"False" or "Yes"/"No". Values that cannot be properly cast cause a ``ValueError`` to be raised. Once parsed, the querystring arguments and keyword arguments are passed to the ``ConnectionPool``'s class initializer. In the case of conflicting arguments, querystring arguments always win. rANrY)rKrk)rrIr url_optionsrYrYrZfrom_urlO s )  zConnectionPool.from_urlNr? cache_factoryrcKs8|pd}t|tr |dkrtd||_||_||_d|_||_|ds*|drd|jddvr6t d|jd}|durMt|t sItd ||_n|jdurY|j |_n t |jd |_| dd| dd|jd d|_|jdurt|_t|_t|_tj|fd |i||dS) Nlrz,"max_connections" must be a positive integer cache_configrrr`z4Client caching is only supported with RESP version 3z#Cache must implement CacheInterfacerr)rHrrrA_connection_kwargsr?r_cache_factoryrfr,r get_cacherpoprr%rr _fork_lock_lockr_rdrP)rTrAr?rrrcrrYrYrZrd sL         zConnectionPool.__init__c CsNddd|jD}d|jjd|jjd|jjd|jjd|d S) Nr2cSr3r4rYr6rYrYrZr: r;z+ConnectionPool.__repr__..r<r=z(z)>)>)rfrcrr?r]r\rA)rT conn_kwargsrYrYrZrA szConnectionPool.__repr__cCrrarrsrYrYrZrc rz ConnectionPool.connection_kwargsrWcCrrarrrYrYrZrc rcCs|jddS)rrN)rcrfrsrYrYrZr} szConnectionPool.get_protocolcCs"d|_g|_t|_t|_dS)Nr)_created_connections_available_connectionsr_in_use_connectionsrrrrsrYrYrZrP s zConnectionPool.resetcCsl|jtkr4|jjdd}|stz|jtkr&|W|jdSW|jdS|jwdS)N)r)rrrracquirer(rPr[)rTacquiredrYrYrZ _checkpid s# zConnectionPool._checkpidrQrRrSrTrc Os||j#z|j}Wn ty|}Ynw|j|Wdn1s-wYz@|z| rI|j durM| sQt dWW|SWW|SWW|St t tfyr||| rnt dYW|Swty||w)zGet a connection from the poolNConnection has dataConnection not ready)rrrr IndexErrormake_connectionraddr~rrrhr)r.rrr}r[rTrWrrXrrYrYrZrY sH     zConnectionPool.get_connectioncCs,|j}t|dd|dd|dddS)z,Return an encoder based on encoding settingsr rr rrF)r r r)rcrrfrxrYrYrZrZA s    zConnectionPool.get_encoderrqcCsb|j|jkr td|jd7_t|j}|jdur)t|jdi||j|jS|jdi|S)zCreate a new connectionzToo many connectionsrNrY) rr?r+dictrcrrrArrxrYrYrZrJ s   zConnectionPool.make_connectionrc Cs||jKz|j|Wnty YWddSw||r=|r.||j ||j t |n | WddSWddS1sUwYdS)z(Releases the connection back to the poolN) rrrrGKeyErrorowns_connectionrrrrMrdispatchr$rrYrYrZr[X s(    "zConnectionPool.releasecCs |j|jkSra)rrrYrYrZrr rzConnectionPool.owns_connectionTr\cCs`||j|rt|j|j}n|j}|D]}|qWddS1s)wYdS)a Disconnects connections in the pool If ``inuse_connections`` is True, disconnect connections that are currently in use, potentially by other threads. Otherwise only disconnect connections that are idle in the pool. N)rrrrrrrTr\ connectionsrrYrYrZru s "zConnectionPool.disconnectcCru)z-Close the pool, disconnecting all connectionsNrPrsrYrYrZrm rxzConnectionPool.closercCs8|jd|i|jD]}||_q |jD]}||_qdS)Nr)rcrkrrr)rTrrrYrYrZr] s   zConnectionPool.set_retryrcsj9jD]!jfddfddjfddfddqjD]q,WddS1s?wYdS)NcsddSr)rrrrY)rrrYrZrM sz1ConnectionPool.re_auth_callback..crOra_mockrQrsrYrZrM rScrKra)rrY)rrYrZrM rNcrOrarrQrsrYrZrM rS)rrrrUrrrrY)rrTrrZr^ s     "zConnectionPool.re_auth_callbackcCrra)rrsrYrYrZrd rHzConnectionPool._get_pool_lockcC0|j |jWdS1swYdSra)rrrsrYrYrZre $z$ConnectionPool._get_free_connectionscCrra)rrrsrYrYrZrf rz&ConnectionPool._get_in_use_connectionsrRcsdS)z Dummy functions, needs to be passed as error callback to retry object. :param error: :return: NrYrvrYrYrZr szConnectionPool._mockrpra)r`r)r`rq)rrr`Nr)+r\r]r^r classmethodrrLrIrrrrrr0rdrArr rrcrr}rPrr8rYrrZrr[rrrrmr3r]r r^rdrerfr,rrYrYrYrZrM: sR2 @  /%  cseZdZdZddeeffdd ZddZdd Ze d gd d d dddZ ddZ dde fddZ ddZddZde fddZZS) BlockingConnectionPoola Thread-safe blocking connection pool:: >>> from redis.client import Redis >>> client = Redis(connection_pool=BlockingConnectionPool()) It performs the same function as the default :py:class:`~redis.ConnectionPool` implementation, in that, it maintains a pool of reusable connections that can be shared by multiple redis clients (safely across threads if required). The difference is that, in the event that a client tries to get a connection from the pool when all of connections are in use, rather than raising a :py:class:`~redis.ConnectionError` (as the default :py:class:`~redis.ConnectionPool` implementation does), it makes the client wait ("blocks") for a specified number of seconds until a connection becomes available. Use ``max_connections`` to increase / decrease the pool size:: >>> pool = BlockingConnectionPool(max_connections=10) Use ``timeout`` to tell it either how many seconds to wait for a connection to become available, or to block forever: >>> # Block forever. >>> pool = BlockingConnectionPool(timeout=None) >>> # Raise a ``ConnectionError`` after five seconds if a connection is >>> # not available. >>> pool = BlockingConnectionPool(timeout=5) 2c s4||_||_d|_d|_tjd||d|dS)NF)rAr?rY) queue_classr_in_maintenance_lockedrrd)rTr?rrArrcrrYrZrd s zBlockingConnectionPool.__init__cCszB|jr |jd|_||j|_ z|jdWn ty%Ynwqg|_ W|jrBz|j Wn t y>Ynwd|_n|jrZz|j Wn t yVYnwd|_wt |_dS)NTF)rrrrrr?pool put_nowaitr _connectionsr[rrrrrsrYrYrZrP s:     zBlockingConnectionPool.resetcCszJ|jr |jd|_|jdur!t|jdi|j|j|j}n |jdi|j}|j ||W|jrJz|j Wn t yEYnwd|_SS|jrbz|j Wn t y^Ynwd|_w)zMake a fresh connection.TNFrY) rrrrrrrArcrrMr[rrrYrYrZr s6     z&BlockingConnectionPool.make_connectionrQrRrSrTNcOs>|d}zC|jr|jd|_z |jjd|jd}Wn ty(t dw|dur1| }W|jrIz|j Wn t yEYnwd|_n|jraz|j Wn t y]Ynwd|_wz/| z |rpt dWW|St ttfy|| |rt dYW|Swty| |w)a7 Get a connection, blocking for ``self.timeout`` until a connection is available from the pool. If the connection returned is ``None`` then creates a new connection. Because we use a last-in first-out queue, the existing connections (having been returned to the pool after the initial ``None`` values were added) will be returned before ``None`` values. This means we only create new connections when we need to, i.e.: the actual number of connections will only increase in response to demand. NT)blockrzNo connection available.Frr)rrrrrrrfrrr)rr[rr~rr.rrr}rrYrYrZrY' s`       z%BlockingConnectionPool.get_connectioncCs|zl|jr|jd|_||s;||jdW|jr9z|j Wn t y3Ynwd|_dSdS| rC|z|j|Wn t yTYnwW|jroz|j Wn t yiYnwd|_dSdS|jrz|j Wn t yYnwd|_w)z)Releases the connection back to the pool.TNF) rrrrrrrrrr[rrrrrYrYrZr[k sN           zBlockingConnectionPool.releaseTr\cCs|z:|jr|jd|_|r|j}n|}|D]}|qW|jr=z|jWn t y7Ynwd|_dSdS|jrVz|jWn t yRYnwd|_w)zLDisconnects either all connections in the pool or just the free connections.TFN) rrrrrrrerr[rrrYrYrZr s2      z!BlockingConnectionPool.disconnectcCs<|jdd|jjDWdS1swYdS)NcSh|]}|r|qSrYrYr7rrYrYrZ z?BlockingConnectionPool._get_free_connections..)rrqueuersrYrYrZre s$z,BlockingConnectionPool._get_free_connectionscsP|jdd|jjDfdd|jDWdS1s!wYdS)NcSrrYrYrrYrYrZr rzABlockingConnectionPool._get_in_use_connections..csh|]}|vr|qSrYrYrconnections_in_queuerYrZr s)rrrrrsrYrrZrf s  $z.BlockingConnectionPool._get_in_use_connectionsin_maintenancecCs ||_dS)a Sets a flag that this Blocking ConnectionPool is in maintenance mode. This is used to prevent new connections from being created while we are in maintenance mode. The pool will be in maintenance mode only when we are processing a MOVING notification. N)r)rTrrYrYrZset_in_maintenance s z)BlockingConnectionPool.set_in_maintenancerar)r\r]r^rrrrdrPrr8rYr[rrrerfrrrYrYrrZr s(## ?" r)rr"rrrQrryrDabcrr itertoolsrrrrrtypingrr r r r r rrrr urllib.parserrr redis.cacherrrrrr_parsersrrrr auth.tokenr backoffr! credentialsr"r#eventr$r% exceptionsr&r'r(r)r*r+r,r-r.maint_notificationsr/r0r1r2rr3utilsr4r5r6r7r8r9r:r;r<rr=rNrgrlrirer+rr__annotations__rBrCr_rqrrrrrr-r7r8r>rrr!rFrKrLrOr_rMrrYrYrYrZs  0   , , 3gKSN9# 9-Z