Low Level APIs¶
Connection¶
BaseParser
¶
Plain Python parsing class
parse_error(self, response)
¶
Parse an error response
Source code in aioredis/connection.py
def parse_error(self, response: str) -> ResponseError:
"""Parse an error response"""
error_code = response.split(" ")[0]
if error_code in self.EXCEPTION_CLASSES:
response = response[len(error_code) + 1 :]
exception_class_or_dict = self.EXCEPTION_CLASSES[error_code]
if isinstance(exception_class_or_dict, dict):
exception_class = exception_class_or_dict.get(response, ResponseError)
else:
exception_class = exception_class_or_dict
return exception_class(response)
return ResponseError(response)
BlockingConnectionPool (ConnectionPool)
¶
Thread-safe blocking connection pool::
>>> from aioredis.client import Redis
>>> client = Redis(connection_pool=BlockingConnectionPool())
It performs the same function as the default
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 class:~redis.ConnectionError
(as the default
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)
disconnect(self, inuse_connections=True)
async
¶
Disconnects all connections in the pool.
Source code in aioredis/connection.py
async def disconnect(self, inuse_connections: bool = True):
"""Disconnects all connections in the pool."""
self._checkpid()
async with self._lock:
resp = await asyncio.gather(
*(connection.disconnect() for connection in self._connections),
return_exceptions=True,
)
exc = next((r for r in resp if isinstance(r, BaseException)), None)
if exc:
raise exc
get_connection(self, command_name, *keys, **options)
async
¶
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.
Source code in aioredis/connection.py
async def get_connection(self, command_name, *keys, **options):
"""
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.
"""
# Make sure we haven't changed process.
self._checkpid()
# Try and get a connection from the pool. If one isn't available within
# self.timeout then raise a ``ConnectionError``.
connection = None
try:
async with async_timeout.timeout(self.timeout):
connection = await self.pool.get()
except (asyncio.QueueEmpty, asyncio.TimeoutError):
# Note that this is not caught by the redis client and will be
# raised unless handled by application code. If you want never to
raise ConnectionError("No connection available.")
# If the ``connection`` is actually ``None`` then that's a cue to make
# a new connection to add to the pool.
if connection is None:
connection = self.make_connection()
try:
# ensure this connection is connected to Redis
await connection.connect()
# connections that the pool provides should be ready to send
# a command. if not, the connection was either returned to the
# pool before all data has been read or the socket has been
# closed. either way, reconnect and verify everything is good.
try:
if await connection.can_read():
raise ConnectionError("Connection has data") from None
except ConnectionError:
await connection.disconnect()
await connection.connect()
if await connection.can_read():
raise ConnectionError("Connection not ready") from None
except BaseException:
# release the connection back to the pool so that we don't leak it
await self.release(connection)
raise
return connection
make_connection(self)
¶
Make a fresh connection.
Source code in aioredis/connection.py
def make_connection(self):
"""Make a fresh connection."""
connection = self.connection_class(**self.connection_kwargs)
self._connections.append(connection)
return connection
release(self, connection)
async
¶
Releases the connection back to the pool.
Source code in aioredis/connection.py
async def release(self, connection: Connection):
"""Releases the connection back to the pool."""
# Make sure we haven't changed process.
self._checkpid()
if not self.owns_connection(connection):
# pool doesn't own this connection. do not add it back
# to the pool. instead add a None value which is a placeholder
# that will cause the pool to recreate the connection if
# its needed.
await connection.disconnect()
self.pool.put_nowait(None)
return
# Put the connection back into the pool.
try:
self.pool.put_nowait(connection)
except asyncio.QueueFull:
# perhaps the pool has been reset() after a fork? regardless,
# we don't want this connection
pass
Connection
¶
Manages TCP communication to and from a Redis server
can_read(self, timeout=0)
async
¶
Poll the socket to see if there’s data that can be read.
Source code in aioredis/connection.py
async def can_read(self, timeout: float = 0):
"""Poll the socket to see if there's data that can be read."""
if not self.is_connected:
await self.connect()
return await self._parser.can_read(timeout)
check_health(self)
async
¶
Check the health of the connection with a PING/PONG
Source code in aioredis/connection.py
async def check_health(self):
"""Check the health of the connection with a PING/PONG"""
if (
self.health_check_interval
and asyncio.get_event_loop().time() > self.next_health_check
):
try:
await self.send_command("PING", check_health=False)
if str_if_bytes(await self.read_response()) != "PONG":
raise ConnectionError("Bad response from PING health check")
except (ConnectionError, TimeoutError) as err:
await self.disconnect()
try:
await self.send_command("PING", check_health=False)
if str_if_bytes(await self.read_response()) != "PONG":
raise ConnectionError(
"Bad response from PING health check"
) from None
except BaseException as err2:
raise err2 from err
connect(self)
async
¶
Connects to the Redis server if not already connected
Source code in aioredis/connection.py
async def connect(self):
"""Connects to the Redis server if not already connected"""
if self.is_connected:
return
try:
await self._connect()
except asyncio.CancelledError:
raise
except (socket.timeout, asyncio.TimeoutError):
raise TimeoutError("Timeout connecting to server")
except OSError as e:
raise ConnectionError(self._error_message(e))
except Exception as exc:
raise ConnectionError(exc) from exc
try:
await self.on_connect()
except RedisError:
# clean up after any error in on_connect
await self.disconnect()
raise
# run any user callbacks. right now the only internal callback
# is for pubsub channel/pattern resubscription
for callback in self._connect_callbacks:
task = callback(self)
if task and inspect.isawaitable(task):
await task
disconnect(self)
async
¶
Disconnects from the Redis server
Source code in aioredis/connection.py
async def disconnect(self):
"""Disconnects from the Redis server"""
try:
async with async_timeout.timeout(self.socket_connect_timeout):
self._parser.on_disconnect()
if not self.is_connected:
return
try:
if os.getpid() == self.pid:
self._writer.close() # type: ignore[union-attr]
# py3.6 doesn't have this method
if hasattr(self._writer, "wait_closed"):
await self._writer.wait_closed() # type: ignore[union-attr]
except OSError:
pass
self._reader = None
self._writer = None
except asyncio.TimeoutError:
raise TimeoutError(
f"Timed out closing connection after {self.socket_connect_timeout}"
) from None
on_connect(self)
async
¶
Initialize the connection, authenticate and select a database
Source code in aioredis/connection.py
async def on_connect(self):
"""Initialize the connection, authenticate and select a database"""
self._parser.on_connect(self)
# if username and/or password are set, authenticate
if self.username or self.password:
auth_args: Union[Tuple[str], Tuple[str, str]]
if self.username:
auth_args = (self.username, self.password or "")
else:
# Mypy bug: https://github.com/python/mypy/issues/10944
auth_args = (self.password or "",)
# avoid checking health here -- PING will fail if we try
# to check the health prior to the AUTH
await self.send_command("AUTH", *auth_args, check_health=False)
try:
auth_response = await self.read_response()
except AuthenticationWrongNumberOfArgsError:
# a username and password were specified but the Redis
# server seems to be < 6.0.0 which expects a single password
# arg. retry auth with just the password.
# https://github.com/andymccurdy/redis-py/issues/1274
await self.send_command("AUTH", self.password, check_health=False)
auth_response = await self.read_response()
if str_if_bytes(auth_response) != "OK":
raise AuthenticationError("Invalid Username or Password")
# if a client_name is given, set it
if self.client_name:
await self.send_command("CLIENT", "SETNAME", self.client_name)
if str_if_bytes(await self.read_response()) != "OK":
raise ConnectionError("Error setting client name")
# if a database is specified, switch to it
if self.db:
await self.send_command("SELECT", self.db)
if str_if_bytes(await self.read_response()) != "OK":
raise ConnectionError("Invalid Database")
pack_command(self, *args)
¶
Pack a series of arguments into the Redis protocol
Source code in aioredis/connection.py
def pack_command(self, *args: EncodableT) -> List[bytes]:
"""Pack a series of arguments into the Redis protocol"""
output = []
# the client might have included 1 or more literal arguments in
# the command name, e.g., 'CONFIG GET'. The Redis server expects these
# arguments to be sent separately, so split the first argument
# manually. These arguments should be bytestrings so that they are
# not encoded.
assert not isinstance(args[0], float)
if isinstance(args[0], str):
args = tuple(args[0].encode().split()) + args[1:]
elif b" " in args[0]:
args = tuple(args[0].split()) + args[1:]
buff = SYM_EMPTY.join((SYM_STAR, str(len(args)).encode(), SYM_CRLF))
buffer_cutoff = self._buffer_cutoff
for arg in map(self.encoder.encode, args):
# to avoid large string mallocs, chunk the command into the
# output list if we're sending large values or memoryviews
arg_length = len(arg)
if (
len(buff) > buffer_cutoff
or arg_length > buffer_cutoff
or isinstance(arg, memoryview)
):
buff = SYM_EMPTY.join(
(buff, SYM_DOLLAR, str(arg_length).encode(), SYM_CRLF)
)
output.append(buff)
output.append(arg)
buff = SYM_CRLF
else:
buff = SYM_EMPTY.join(
(
buff,
SYM_DOLLAR,
str(arg_length).encode(),
SYM_CRLF,
arg,
SYM_CRLF,
)
)
output.append(buff)
return output
pack_commands(self, commands)
¶
Pack multiple commands into the Redis protocol
Source code in aioredis/connection.py
def pack_commands(self, commands: Iterable[Iterable[EncodableT]]) -> List[bytes]:
"""Pack multiple commands into the Redis protocol"""
output: List[bytes] = []
pieces: List[bytes] = []
buffer_length = 0
buffer_cutoff = self._buffer_cutoff
for cmd in commands:
for chunk in self.pack_command(*cmd):
chunklen = len(chunk)
if (
buffer_length > buffer_cutoff
or chunklen > buffer_cutoff
or isinstance(chunk, memoryview)
):
output.append(SYM_EMPTY.join(pieces))
buffer_length = 0
pieces = []
if chunklen > buffer_cutoff or isinstance(chunk, memoryview):
output.append(chunk)
else:
pieces.append(chunk)
buffer_length += chunklen
if pieces:
output.append(SYM_EMPTY.join(pieces))
return output
read_response(self)
async
¶
Read the response from a previously sent command
Source code in aioredis/connection.py
async def read_response(self):
"""Read the response from a previously sent command"""
try:
async with self._lock:
async with async_timeout.timeout(self.socket_timeout):
response = await self._parser.read_response()
except asyncio.TimeoutError:
await self.disconnect()
raise TimeoutError(f"Timeout reading from {self.host}:{self.port}")
except OSError as e:
await self.disconnect()
raise ConnectionError(
f"Error while reading from {self.host}:{self.port} : {e.args}"
)
except BaseException:
await self.disconnect()
raise
if self.health_check_interval:
self.next_health_check = (
asyncio.get_event_loop().time() + self.health_check_interval
)
if isinstance(response, ResponseError):
raise response from None
return response
send_command(self, *args, **kwargs)
async
¶
Pack and send a command to the Redis server
Source code in aioredis/connection.py
async def send_command(self, *args, **kwargs):
"""Pack and send a command to the Redis server"""
if not self.is_connected:
await self.connect()
await self.send_packed_command(
self.pack_command(*args), check_health=kwargs.get("check_health", True)
)
send_packed_command(self, command, check_health=True)
async
¶
Send an already packed command to the Redis server
Source code in aioredis/connection.py
async def send_packed_command(
self,
command: Union[bytes, str, Iterable[bytes]],
check_health: bool = True,
):
"""Send an already packed command to the Redis server"""
if not self._writer:
await self.connect()
# guard against health check recursion
if check_health:
await self.check_health()
try:
if isinstance(command, str):
command = command.encode()
if isinstance(command, bytes):
command = [command]
await asyncio.wait_for(
self._send_packed_command(command),
self.socket_timeout,
)
except asyncio.TimeoutError:
await self.disconnect()
raise TimeoutError("Timeout writing to socket") from None
except OSError as e:
await self.disconnect()
if len(e.args) == 1:
err_no, errmsg = "UNKNOWN", e.args[0]
else:
err_no = e.args[0]
errmsg = e.args[1]
raise ConnectionError(
f"Error {err_no} while writing to socket. {errmsg}."
) from e
except BaseException:
await self.disconnect()
raise
ConnectionPool
¶
Create a connection pool. If max_connections
is set, then this
object raises class:~redis.ConnectionError
when the pool’s
limit is reached.
By default, TCP connections are created unless connection_class
is specified. Use class:~redis.UnixDomainSocketConnection
for
unix sockets.
Any additional keyword arguments are passed to the constructor of
connection_class
.
disconnect(self, inuse_connections=True)
async
¶
Disconnects connections in the pool
If inuse_connections
is True, disconnect connections that are
current in use, potentially by other tasks. Otherwise only disconnect
connections that are idle in the pool.
Source code in aioredis/connection.py
async def disconnect(self, inuse_connections: bool = True):
"""
Disconnects connections in the pool
If ``inuse_connections`` is True, disconnect connections that are
current in use, potentially by other tasks. Otherwise only disconnect
connections that are idle in the pool.
"""
self._checkpid()
async with self._lock:
if inuse_connections:
connections: Iterable[Connection] = chain(
self._available_connections, self._in_use_connections
)
else:
connections = self._available_connections
resp = await asyncio.gather(
*(connection.disconnect() for connection in connections),
return_exceptions=True,
)
exc = next((r for r in resp if isinstance(r, BaseException)), None)
if exc:
raise exc
from_url(url, **kwargs)
classmethod
¶
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]:[password]]@/path/to/socket.sock?db=0
Three URL schemes are supported:
redis://
creates a TCP socket connection. See more at: https://www.iana.org/assignments/uri-schemes/prov/redisrediss://
creates a SSL wrapped TCP socket connection. See more at: https://www.iana.org/assignments/uri-schemes/prov/redissunix://
: 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.
Source code in aioredis/connection.py
@classmethod
def from_url(cls: Type[_CP], url: str, **kwargs) -> _CP:
"""
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]:[password]]@/path/to/socket.sock?db=0
Three URL schemes are supported:
- `redis://` creates a TCP socket connection. See more at:
<https://www.iana.org/assignments/uri-schemes/prov/redis>
- `rediss://` creates a SSL wrapped TCP socket connection. See more at:
<https://www.iana.org/assignments/uri-schemes/prov/rediss>
- ``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.
"""
url_options = parse_url(url)
kwargs.update(url_options)
return cls(**kwargs)
get_connection(self, command_name, *keys, **options)
async
¶
Get a connection from the pool
Source code in aioredis/connection.py
async def get_connection(self, command_name, *keys, **options):
"""Get a connection from the pool"""
self._checkpid()
async with self._lock:
try:
connection = self._available_connections.pop()
except IndexError:
connection = self.make_connection()
self._in_use_connections.add(connection)
try:
# ensure this connection is connected to Redis
await connection.connect()
# connections that the pool provides should be ready to send
# a command. if not, the connection was either returned to the
# pool before all data has been read or the socket has been
# closed. either way, reconnect and verify everything is good.
try:
if await connection.can_read():
raise ConnectionError("Connection has data") from None
except ConnectionError:
await connection.disconnect()
await connection.connect()
if await connection.can_read():
raise ConnectionError("Connection not ready") from None
except BaseException:
# release the connection back to the pool so that we don't
# leak it
await self.release(connection)
raise
return connection
get_encoder(self)
¶
Return an encoder based on encoding settings
Source code in aioredis/connection.py
def get_encoder(self):
"""Return an encoder based on encoding settings"""
kwargs = self.connection_kwargs
return self.encoder_class(
encoding=kwargs.get("encoding", "utf-8"),
encoding_errors=kwargs.get("encoding_errors", "strict"),
decode_responses=kwargs.get("decode_responses", False),
)
make_connection(self)
¶
Create a new connection
Source code in aioredis/connection.py
def make_connection(self):
"""Create a new connection"""
if self._created_connections >= self.max_connections:
raise ConnectionError("Too many connections")
self._created_connections += 1
return self.connection_class(**self.connection_kwargs)
release(self, connection)
async
¶
Releases the connection back to the pool
Source code in aioredis/connection.py
async def release(self, connection: Connection):
"""Releases the connection back to the pool"""
self._checkpid()
async with self._lock:
try:
self._in_use_connections.remove(connection)
except KeyError:
# Gracefully fail when a connection is returned to this pool
# that the pool doesn't actually own
pass
if self.owns_connection(connection):
self._available_connections.append(connection)
else:
# pool doesn't own this connection. do not add it back
# to the pool and decrement the count so that another
# connection can take its place if needed
self._created_connections -= 1
await connection.disconnect()
return
DefaultParser (BaseParser)
¶
Plain Python parsing class
on_connect(self, connection)
¶
Called when the stream connects
Source code in aioredis/connection.py
def on_connect(self, connection: "Connection"):
"""Called when the stream connects"""
self._stream = connection._reader
if self._stream is None:
raise RedisError("Buffer is closed.")
self._buffer = SocketBuffer(
self._stream, self._read_size, connection.socket_timeout
)
self.encoder = connection.encoder
on_disconnect(self)
¶
Called when the stream disconnects
Source code in aioredis/connection.py
def on_disconnect(self):
"""Called when the stream disconnects"""
if self._stream is not None:
self._stream = None
if self._buffer is not None:
self._buffer.close()
self._buffer = None
self.encoder = None
Encoder
¶
Encode strings to bytes-like and decode bytes-like to strings
decode(self, value, force=False)
¶
Return a unicode string from the bytes-like representation
Source code in aioredis/connection.py
def decode(self, value: EncodableT, force=False) -> EncodableT:
"""Return a unicode string from the bytes-like representation"""
if self.decode_responses or force:
if isinstance(value, memoryview):
return value.tobytes().decode(self.encoding, self.encoding_errors)
if isinstance(value, bytes):
return value.decode(self.encoding, self.encoding_errors)
return value
encode(self, value)
¶
Return a bytestring or bytes-like representation of the value
Source code in aioredis/connection.py
def encode(self, value: EncodableT) -> EncodedT:
"""Return a bytestring or bytes-like representation of the value"""
if isinstance(value, (bytes, memoryview)):
return value
if isinstance(value, bool):
# special case bool since it is a subclass of int
raise DataError(
"Invalid input of type: 'bool'. "
"Convert to a bytes, string, int or float first."
)
if isinstance(value, (int, float)):
return repr(value).encode()
if not isinstance(value, str):
# a value we don't know how to deal with. throw an error
typename = value.__class__.__name__ # type: ignore[unreachable]
raise DataError(
f"Invalid input of type: {typename!r}. "
"Convert to a bytes, string, int or float first."
)
return value.encode(self.encoding, self.encoding_errors)
HiredisParser (BaseParser)
¶
Parser class for connections using Hiredis
PythonParser (BaseParser)
¶
Plain Python parsing class
on_connect(self, connection)
¶
Called when the stream connects
Source code in aioredis/connection.py
def on_connect(self, connection: "Connection"):
"""Called when the stream connects"""
self._stream = connection._reader
if self._stream is None:
raise RedisError("Buffer is closed.")
self._buffer = SocketBuffer(
self._stream, self._read_size, connection.socket_timeout
)
self.encoder = connection.encoder
on_disconnect(self)
¶
Called when the stream disconnects
Source code in aioredis/connection.py
def on_disconnect(self):
"""Called when the stream disconnects"""
if self._stream is not None:
self._stream = None
if self._buffer is not None:
self._buffer.close()
self._buffer = None
self.encoder = None
SocketBuffer
¶
Async-friendly re-impl of redis-py’s SocketBuffer.
TODO: We’re currently passing through two buffers, the asyncio.StreamReader and this. I imagine we can reduce the layers here while maintaining compliance with prior art.
Utils¶
from_url(url, **kwargs)
¶
Returns an active Redis client generated from the given database URL.
Will attempt to extract the database id from the path url fragment, if none is provided.
Source code in aioredis/utils.py
def from_url(url, **kwargs):
"""
Returns an active Redis client generated from the given database URL.
Will attempt to extract the database id from the path url fragment, if
none is provided.
"""
from aioredis.client import Redis
return Redis.from_url(url, **kwargs)