Skip to content

Commit 7c856ed

Browse files
YarmolukYarmoluk
committed
fix(connection): drop sync pool plumbing and compare sync-only kwargs to their own defaults
The sync pool injects maintenance-notification keywords (the pool handler, maint_notifications_config and the orig_* copies of host and timeouts) for its own connections. Older async classes don't accept them, so on redis-py 7.4 with RESP3, and on 8.0 for every client, the conversion raised naming settings the user never touched. Those keywords describe the sync pool, not the user's settings, and the async pool computes its own, so they are never carried over; maint_notifications_config is dropped unless the user enabled notifications explicitly. A remaining sync-only keyword now counts as unset when it still holds the sync class's own declared default, found by inspect.signature, rather than when it is None or False. That stops 0 being treated as unset and covers keywords whose default is not falsy. Also trims the SemanticRouter.aroute caveat to Sentinel and Cluster now that #784 is in, notes that a custom SSLConnection subclass's behaviour is not carried over, and folds the tests into two parametrised ones with RESP3 and unix-URL cases.
1 parent c5070c6 commit 7c856ed

3 files changed

Lines changed: 155 additions & 93 deletions

File tree

‎redisvl/extensions/router/semantic.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -723,8 +723,8 @@ async def aroute(
723723
it returns the same match as ``router(statement)``.
724724
725725
Pass ``redis_url`` rather than a client to the router if you need
726-
async routing over SSL, a unix socket, Sentinel or Redis Cluster: those
727-
clients cannot be converted to their async form. With a vectorizer that
726+
async routing over Sentinel or Redis Cluster: those clients cannot be
727+
converted to their async form. With a vectorizer that
728728
has no native ``aembed``, such as the default ``HFTextVectorizer``,
729729
embedding still blocks the loop; only the Redis call is non-blocking.
730730
The async client is bound to the event loop that made the first call,

‎redisvl/redis/connection.py‎

Lines changed: 66 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -306,48 +306,88 @@ async def _aidentify_client(
306306
logger.debug(f"CLIENT SETINFO was not applied, continuing without it: {e}")
307307

308308

309-
def _accepted_init_kwargs(cls: type) -> set[str]:
310-
"""Return the keyword arguments ``cls(...)`` accepts, walking its MRO.
309+
def _init_parameters(cls: type) -> dict[str, inspect.Parameter]:
310+
"""Return the keyword parameters ``cls(...)`` accepts, walking its MRO.
311311
312312
redis-py's connection classes each declare their own keywords and pass the
313313
rest up with ``**kwargs``, so the full set is the union of every
314-
``__init__`` in the chain up to the first one without ``**kwargs``.
314+
``__init__`` in the chain up to the first one without ``**kwargs``. That
315+
stop is deliberate: a subclass whose ``__init__`` takes no ``**kwargs``
316+
can't be given its parents' keywords, so they aren't accepted either.
315317
"""
316-
accepted: set[str] = set()
318+
params: dict[str, inspect.Parameter] = {}
317319
for klass in cls.__mro__:
318320
init = klass.__dict__.get("__init__")
319321
if init is None:
320322
continue
321-
params = inspect.signature(init).parameters
322-
accepted.update(
323-
name
324-
for name, param in params.items()
325-
if name != "self"
326-
and param.kind in (param.POSITIONAL_OR_KEYWORD, param.KEYWORD_ONLY)
327-
)
328-
if not any(p.kind is p.VAR_KEYWORD for p in params.values()):
323+
signature = inspect.signature(init).parameters
324+
for name, param in signature.items():
325+
if name == "self" or name in params:
326+
continue
327+
if param.kind in (param.POSITIONAL_OR_KEYWORD, param.KEYWORD_ONLY):
328+
params[name] = param
329+
if not any(p.kind is p.VAR_KEYWORD for p in signature.values()):
329330
break
330-
return accepted
331+
return params
332+
333+
334+
# Keywords the sync pool computes for its own connections (redis-py >= 7.4
335+
# fills them in for maintenance notifications). They describe the sync pool,
336+
# not the user's settings, and the async pool computes its own, so they are
337+
# never carried over.
338+
_SYNC_POOL_PLUMBING = frozenset(
339+
{
340+
"maint_notifications_pool_handler",
341+
"orig_host_address",
342+
"orig_socket_timeout",
343+
"orig_socket_connect_timeout",
344+
}
345+
)
346+
347+
348+
def _is_unset(
349+
name: str, value: Any, sync_defaults: dict[str, inspect.Parameter]
350+
) -> bool:
351+
"""Whether ``value`` is what the sync connection class would have used anyway.
352+
353+
The comparison is against the sync class's own declared default, not
354+
against ``None``/``False``, so ``0`` is a setting and a keyword whose
355+
default is ``True`` is unset only when it is still ``True``.
356+
"""
357+
if name == "maint_notifications_config":
358+
# The sync pool injects a config in "auto" mode on its own; only a
359+
# user who turned notifications on explicitly would lose anything.
360+
return getattr(value, "enabled", None) is not True
361+
param = sync_defaults.get(name)
362+
if param is None or param.default is param.empty:
363+
return value is None
364+
default = param.default
365+
return value is default or (type(value) is type(default) and value == default)
331366

332367

333368
def _async_connection_kwargs(
334369
connection_kwargs: Mapping[str, Any],
335370
connection_class: type[AsyncAbstractConnection],
371+
sync_connection_class: type,
336372
) -> dict[str, Any]:
337373
"""Translate a sync pool's ``connection_kwargs`` for an async connection class.
338374
339-
Keywords the async class does not accept are dropped when they hold their
340-
default (``None`` or ``False``). One that is set, such as
341-
``ssl_validate_ocsp=True``, is a setting the async client cannot honour,
342-
so it raises instead of being silently lost.
375+
Keywords the async class does not accept are dropped when they are unset,
376+
meaning they still hold the sync class's own default, or when they are
377+
pool plumbing the async pool computes for itself. One that the user set,
378+
such as ``ssl_validate_ocsp=True``, is a setting the async client cannot
379+
honour, so it raises instead of being silently lost.
343380
"""
344-
accepted = _accepted_init_kwargs(connection_class)
381+
accepted = _init_parameters(connection_class)
382+
sync_defaults = _init_parameters(sync_connection_class)
345383
kwargs: dict[str, Any] = {}
346384
lost: list[str] = []
347385
for name, value in connection_kwargs.items():
386+
if name in _SYNC_POOL_PLUMBING:
387+
continue
348388
if name in accepted:
349389
kwargs[name] = value
350-
elif value not in (None, False):
390+
elif not _is_unset(name, value, sync_defaults):
351391
lost.append(name)
352392
if lost:
353393
raise ValueError(
@@ -935,9 +975,10 @@ def sync_to_async_redis(
935975
"""Convert a synchronous Redis client to an asynchronous one.
936976
937977
The async client is built from the sync client's connection settings:
938-
TLS and unix-socket connections (including subclasses) map to their
939-
async connection classes, and keywords that only the sync classes
940-
accept are dropped when they hold their default.
978+
TLS and unix-socket connections map to redis-py's stock async
979+
connection classes, so the behaviour of a custom ``SSLConnection``
980+
subclass is not carried over. Keywords that only the sync classes
981+
accept are dropped when unset.
941982
942983
Raises:
943984
ValueError: For a ``RedisCluster`` client, a Sentinel-managed
@@ -974,7 +1015,9 @@ def sync_to_async_redis(
9741015
return AsyncRedis.from_pool(
9751016
AsyncConnectionPool(
9761017
connection_class=connection_class,
977-
**_async_connection_kwargs(pool.connection_kwargs, connection_class),
1018+
**_async_connection_kwargs(
1019+
pool.connection_kwargs, connection_class, pool.connection_class
1020+
),
9781021
)
9791022
)
9801023

‎tests/unit/test_sync_to_async_redis.py‎

Lines changed: 87 additions & 68 deletions
Original file line numberDiff line numberDiff line change
@@ -14,85 +14,104 @@
1414
from redis.connection import ConnectionPool, SSLConnection
1515
from redis.sentinel import Sentinel
1616

17-
from redisvl.redis.connection import RedisConnectionFactory
17+
from redisvl.redis.connection import (
18+
RedisConnectionFactory,
19+
_async_connection_kwargs,
20+
_init_parameters,
21+
)
1822

1923

20-
def _first_connection(async_client):
21-
return async_client.connection_pool.make_connection()
24+
class _TLS(SSLConnection):
25+
pass
26+
27+
28+
@pytest.mark.parametrize(
29+
"client, expected, settings",
30+
[
31+
(
32+
Redis(host="h", port=6380, db=3, decode_responses=True),
33+
AsyncConnection,
34+
{"host": "h", "port": 6380, "db": 3, "decode_responses": True},
35+
),
36+
# RESP3 makes the sync pool inject maintenance-notification plumbing
37+
# that older async classes don't accept (redis-py 7.4 and 8.0).
38+
(
39+
Redis(host="h", protocol=3),
40+
AsyncConnection,
41+
{"host": "h", "protocol": 3},
42+
),
43+
(
44+
Redis(host="h", ssl=True, password="pw", ssl_check_hostname=True),
45+
AsyncSSLConnection,
46+
{"host": "h", "password": "pw", "ssl_check_hostname": True},
47+
),
48+
(
49+
Redis(connection_pool=ConnectionPool(connection_class=_TLS, host="h")),
50+
AsyncSSLConnection,
51+
{"host": "h"},
52+
),
53+
(
54+
Redis(unix_socket_path="/tmp/redis.sock", db=2),
55+
AsyncUnixDomainSocketConnection,
56+
{"path": "/tmp/redis.sock", "db": 2},
57+
),
58+
# The URL form sets connection_class rather than unix_socket_path.
59+
(
60+
Redis.from_url("unix:///tmp/redis.sock"),
61+
AsyncUnixDomainSocketConnection,
62+
{"path": "/tmp/redis.sock"},
63+
),
64+
],
65+
ids=["plain", "resp3", "tls", "tls-subclass", "unix", "unix-url"],
66+
)
67+
def test_converts_to_the_matching_async_connection(client, expected, settings):
68+
pool = RedisConnectionFactory.sync_to_async_redis(client).connection_pool
2269

70+
assert type(pool.make_connection()) is expected
71+
assert settings.items() <= pool.connection_kwargs.items()
2372

24-
def test_plain_client_keeps_its_settings():
25-
client = Redis(host="example.com", port=6380, db=3, decode_responses=True)
2673

27-
async_client = RedisConnectionFactory.sync_to_async_redis(client)
28-
conn = _first_connection(async_client)
74+
@pytest.mark.parametrize(
75+
"client, match",
76+
[
77+
(Sentinel([("localhost", 26379)]).master_for("mymaster"), "Sentinel-managed"),
78+
(Redis(host="h", ssl=True, ssl_validate_ocsp=True), "ssl_validate_ocsp"),
79+
],
80+
ids=["sentinel", "sync-only-setting"],
81+
)
82+
def test_unconvertible_client_raises(client, match):
83+
with pytest.raises(ValueError, match=match):
84+
RedisConnectionFactory.sync_to_async_redis(client)
2985

30-
assert isinstance(conn, AsyncConnection)
31-
assert (conn.host, conn.port, conn.db) == ("example.com", 6380, 3)
32-
assert async_client.connection_pool.connection_kwargs["decode_responses"] is True
3386

87+
def test_sync_only_keyword_is_unset_only_at_its_own_default():
88+
# ssl_validate_ocsp_stapled defaults to False on the sync class; 0 is a
89+
# value the user chose, not the default, so it is not silently dropped.
90+
assert "ssl_validate_ocsp_stapled" not in _init_parameters(AsyncSSLConnection)
3491

35-
def test_tls_keyword_client_converts_and_keeps_tls():
36-
client = Redis(
37-
host="example.com",
38-
ssl=True,
39-
password="secret",
40-
ssl_cert_reqs="required",
41-
ssl_check_hostname=True,
92+
kwargs = _async_connection_kwargs(
93+
{"host": "h", "ssl_validate_ocsp_stapled": False},
94+
AsyncSSLConnection,
95+
SSLConnection,
4296
)
97+
assert kwargs == {"host": "h"}
4398

44-
async_client = RedisConnectionFactory.sync_to_async_redis(client)
45-
conn = _first_connection(async_client)
46-
47-
assert isinstance(conn, AsyncSSLConnection)
48-
assert conn.host == "example.com"
49-
assert conn.ssl_context.check_hostname is True
50-
assert async_client.connection_pool.connection_kwargs["password"] == "secret"
51-
52-
53-
def test_tls_url_client_still_converts():
54-
client = Redis.from_url("rediss://:[email protected]:6380/1")
55-
56-
conn = _first_connection(RedisConnectionFactory.sync_to_async_redis(client))
57-
58-
assert isinstance(conn, AsyncSSLConnection)
59-
assert conn.db == 1
60-
61-
62-
def test_ssl_connection_subclass_gets_a_tls_async_connection():
63-
class CustomSSLConnection(SSLConnection):
64-
pass
65-
66-
client = Redis(
67-
connection_pool=ConnectionPool(
68-
connection_class=CustomSSLConnection, host="example.com", password="pw"
99+
with pytest.raises(ValueError, match="ssl_validate_ocsp_stapled"):
100+
_async_connection_kwargs(
101+
{"host": "h", "ssl_validate_ocsp_stapled": 0},
102+
AsyncSSLConnection,
103+
SSLConnection,
69104
)
70-
)
71-
72-
conn = _first_connection(RedisConnectionFactory.sync_to_async_redis(client))
73-
74-
assert isinstance(conn, AsyncSSLConnection)
75-
76-
77-
def test_unix_socket_client_converts():
78-
client = Redis(unix_socket_path="/tmp/redis.sock", db=2)
79-
80-
conn = _first_connection(RedisConnectionFactory.sync_to_async_redis(client))
81105

82-
assert isinstance(conn, AsyncUnixDomainSocketConnection)
83-
assert conn.path == "/tmp/redis.sock"
84-
assert conn.db == 2
85106

107+
def test_explicitly_enabled_maintenance_notifications_are_not_dropped():
108+
maint = pytest.importorskip("redis.maint_notifications")
109+
config = maint.MaintNotificationsConfig(enabled=True)
110+
client = Redis(host="h", protocol=3, maint_notifications_config=config)
86111

87-
def test_sentinel_managed_client_is_rejected_clearly():
88-
client = Sentinel([("localhost", 26379)]).master_for("mymaster")
89-
90-
with pytest.raises(ValueError, match="Sentinel-managed"):
91-
RedisConnectionFactory.sync_to_async_redis(client)
92-
93-
94-
def test_sync_only_setting_that_is_set_is_not_silently_dropped():
95-
client = Redis(host="example.com", ssl=True, ssl_validate_ocsp=True)
96-
97-
with pytest.raises(ValueError, match="ssl_validate_ocsp"):
98-
RedisConnectionFactory.sync_to_async_redis(client)
112+
if "maint_notifications_config" in _init_parameters(AsyncConnection):
113+
pool = RedisConnectionFactory.sync_to_async_redis(client).connection_pool
114+
assert pool.connection_kwargs["maint_notifications_config"] is config
115+
else:
116+
with pytest.raises(ValueError, match="maint_notifications_config"):
117+
RedisConnectionFactory.sync_to_async_redis(client)

0 commit comments

Comments
 (0)