diff --git a/.github/workflows/docs.yaml b/.github/workflows/docs.yaml index 9f650c91..7b7309b7 100644 --- a/.github/workflows/docs.yaml +++ b/.github/workflows/docs.yaml @@ -43,7 +43,7 @@ jobs: make build-docs - name: upload docs - uses: actions/upload-artifact@v6 + uses: actions/upload-artifact@v7 with: name: valkey-py-docs path: | diff --git a/.github/workflows/integration.yaml b/.github/workflows/integration.yaml index 96ee9771..5a184da4 100644 --- a/.github/workflows/integration.yaml +++ b/.github/workflows/integration.yaml @@ -133,7 +133,7 @@ jobs: make ${{matrix.test-type}}-tests USE_UVLOOP=1 PROTOCOL=${{ matrix.protocol-version }} fi - - uses: actions/upload-artifact@v6 + - uses: actions/upload-artifact@v7 if: success() || failure() with: name: pytest-results-${{matrix.test-type}}-${{matrix.connection-type}}-${{matrix.python-version}}-RESP${{ matrix.protocol-version }} @@ -188,5 +188,7 @@ jobs: python-version: ${{ matrix.python-version }} cache: 'pip' - name: install from pip + env: + PIP_DISABLE_PIP_VERSION_CHECK: "1" run: | pip install --quiet git+${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}.git@${GITHUB_SHA} diff --git a/.github/workflows/spellcheck.yml b/.github/workflows/spellcheck.yml index dbcb5ba7..130cd293 100644 --- a/.github/workflows/spellcheck.yml +++ b/.github/workflows/spellcheck.yml @@ -8,7 +8,7 @@ jobs: - name: Checkout uses: actions/checkout@v6 - name: Check Spelling - uses: rojopolis/spellcheck-github-actions@0.56.0 + uses: rojopolis/spellcheck-github-actions@0.58.0 with: config_path: .github/spellcheck-settings.yml task_name: Markdown diff --git a/README.md b/README.md index a9022591..8cd2546e 100644 --- a/README.md +++ b/README.md @@ -39,12 +39,12 @@ $ pip install "valkey[libvalkey]" ### Basic Example ``` python ->>> import valkey ->>> r = valkey.Valkey(host='localhost', port=6379, db=0) ->>> r.set('foo', 'bar') -True ->>> r.get('foo') -b'bar' +import valkey +r = valkey.Valkey(host='localhost', port=6379, db=0) +r.set('foo', 'bar') +# True +r.get('foo') +# b'bar' ``` The above code connects to localhost on port 6379, sets a value in Redis, and retrieves it. All responses are returned as bytes in Python, to receive decoded strings, set *decode_responses=True*. For this, and more connection options, see [these examples](https://valkey-py.readthedocs.io/en/latest/examples.html). @@ -54,20 +54,20 @@ The above code connects to localhost on port 6379, sets a value in Redis, and re You are encouraged to use the new class names, but to allow for a smooth transition alias are available: ``` python ->>> import valkey as redis ->>> r = redis.Redis(host='localhost', port=6379, db=0) ->>> r.set('foo', 'bar') -True ->>> r.get('foo') -b'bar' +import valkey as redis +r = redis.Redis(host='localhost', port=6379, db=0) +r.set('foo', 'bar') +# True +r.get('foo') +# b'bar' ``` #### RESP3 Support To enable support for RESP3 change your connection object to include *protocol=3* ``` python ->>> import valkey ->>> r = valkey.Valkey(host='localhost', port=6379, db=0, protocol=3) +import valkey +r = valkey.Valkey(host='localhost', port=6379, db=0, protocol=3) ``` ### Connection Pools @@ -75,8 +75,8 @@ To enable support for RESP3 change your connection object to include *protocol=3 By default, valkey-py uses a connection pool to manage connections. Each instance of a Valkey class receives its own connection pool. You can however define your own [valkey.ConnectionPool](https://valkey-py.readthedocs.io/en/latest/connections.html#connection-pools). ``` python ->>> pool = valkey.ConnectionPool(host='localhost', port=6379, db=0) ->>> r = valkey.Valkey(connection_pool=pool) +pool = valkey.ConnectionPool(host='localhost', port=6379, db=0) +r = valkey.Valkey(connection_pool=pool) ``` Alternatively, you might want to look at [Async connections](https://valkey-py.readthedocs.io/en/latest/examples/asyncio_examples.html), or [Cluster connections](https://valkey-py.readthedocs.io/en/latest/connections.html#cluster-client), or even [Async Cluster connections](https://valkey-py.readthedocs.io/en/latest/connections.html#async-cluster-client). @@ -117,12 +117,12 @@ The following is a basic example of a [Valkey pipeline](https://valkey.io/topics ``` python ->>> pipe = r.pipeline() ->>> pipe.set('foo', 5) ->>> pipe.set('bar', 18.5) ->>> pipe.set('blee', "hello world!") ->>> pipe.execute() -[True, True, True] +pipe = r.pipeline() +pipe.set('foo', 5) +pipe.set('bar', 18.5) +pipe.set('blee', "hello world!") +pipe.execute() +# [True, True, True] ``` ### PubSub @@ -130,11 +130,11 @@ The following is a basic example of a [Valkey pipeline](https://valkey.io/topics The following example shows how to utilize [Valkey Pub/Sub](https://valkey.io/topics/pubsub/) to subscribe to specific channels. ``` python ->>> r = valkey.Valkey(...) ->>> p = r.pubsub() ->>> p.subscribe('my-first-channel', 'my-second-channel', ...) ->>> p.get_message() -{'pattern': None, 'type': 'subscribe', 'channel': b'my-second-channel', 'data': 1} +r = valkey.Valkey(...) +p = r.pubsub() +p.subscribe('my-first-channel', 'my-second-channel', ...) +p.get_message() +# {'pattern': None, 'type': 'subscribe', 'channel': b'my-second-channel', 'data': 1} ``` diff --git a/tests/test_asyncio/conftest.py b/tests/test_asyncio/conftest.py index c84fe79d..ec17c439 100644 --- a/tests/test_asyncio/conftest.py +++ b/tests/test_asyncio/conftest.py @@ -78,7 +78,7 @@ async def teardown(): await client.aclose() await client.connection_pool.disconnect() else: - if flushdb: + if flushdb and not getattr(client, "_closed", False): try: await client.flushdb(target_nodes="primaries") except valkey.ConnectionError: diff --git a/tests/test_asyncio/test_cluster.py b/tests/test_asyncio/test_cluster.py index 01c515a0..ac879f35 100644 --- a/tests/test_asyncio/test_cluster.py +++ b/tests/test_asyncio/test_cluster.py @@ -319,6 +319,20 @@ async def mock_aclose(): assert called == 1 await cluster.aclose() + async def test_commands_fail_after_aclose(self) -> None: + cluster = await get_mocked_valkey_client(host=default_host, port=default_port) + await cluster.aclose() + + with pytest.raises(ValkeyClusterException, match="ValkeyCluster is closed"): + await cluster.get("a") + + async def test_initialize_fails_after_aclose(self) -> None: + cluster = await get_mocked_valkey_client(host=default_host, port=default_port) + await cluster.aclose() + + with pytest.raises(ValkeyClusterException, match="ValkeyCluster is closed"): + await cluster.initialize() + async def test_startup_nodes(self) -> None: """ Test that it is possible to use startup_nodes diff --git a/valkey/__init__.py b/valkey/__init__.py index 29cca5b1..e74de088 100644 --- a/valkey/__init__.py +++ b/valkey/__init__.py @@ -42,7 +42,7 @@ def int_or_str(value): return value -__version__ = "6.2.0rc1" +__version__ = "6.2.0rc2" VERSION = tuple(map(int_or_str, __version__.split("."))) Redis = Valkey diff --git a/valkey/asyncio/cluster.py b/valkey/asyncio/cluster.py index 172fdfb3..30c3f022 100644 --- a/valkey/asyncio/cluster.py +++ b/valkey/asyncio/cluster.py @@ -225,6 +225,7 @@ def from_url(cls, url: str, **kwargs: Any) -> "ValkeyCluster": return cls(**kwargs) __slots__ = ( + "_closed", "_initialize", "_lock", "cluster_error_retry_attempts", @@ -417,21 +418,24 @@ def __init__( self.command_flags = self.__class__.COMMAND_FLAGS.copy() self.response_callbacks = kwargs["response_callbacks"] self.result_callbacks = self.__class__.RESULT_CALLBACKS.copy() - self.result_callbacks["CLUSTER SLOTS"] = ( - lambda cmd, res, **kwargs: parse_cluster_slots( - list(res.values())[0], **kwargs - ) + self.result_callbacks["CLUSTER SLOTS"] = lambda cmd, res, **kwargs: ( + parse_cluster_slots(list(res.values())[0], **kwargs) ) + self._closed = False self._initialize = True self._lock: Optional[asyncio.Lock] = None async def initialize(self) -> "ValkeyCluster": """Get all nodes from startup nodes & creates connections if not initialized.""" + if self._closed: + raise ValkeyClusterException("ValkeyCluster is closed") if self._initialize: if not self._lock: self._lock = asyncio.Lock() async with self._lock: + if self._closed: + raise ValkeyClusterException("ValkeyCluster is closed") if self._initialize: try: await self.nodes_manager.initialize() @@ -445,8 +449,19 @@ async def initialize(self) -> "ValkeyCluster": raise return self + async def _reset_for_reinitialize(self) -> None: + if not self._initialize: + if not self._lock: + self._lock = asyncio.Lock() + async with self._lock: + if not self._initialize: + self._initialize = True + await self.nodes_manager.aclose() + await self.nodes_manager.aclose("startup_nodes") + async def aclose(self) -> None: """Close all connections & client if initialized.""" + self._closed = True if not self._initialize: if not self._lock: self._lock = asyncio.Lock() @@ -821,13 +836,13 @@ async def _execute_command( self.nodes_manager.startup_nodes.pop(target_node.name, None) # Hard force of reinitialize of the node/slots setup # and try again with the new setup - await self.aclose() + await self._reset_for_reinitialize() raise except ClusterDownError: # ClusterDownError can occur during a failover and to get # self-healed, we will try to reinitialize the cluster layout # and retry executing the command - await self.aclose() + await self._reset_for_reinitialize() await asyncio.sleep(0.25) raise except MovedError as e: @@ -844,7 +859,7 @@ async def _execute_command( self.reinitialize_steps and self.reinitialize_counter % self.reinitialize_steps == 0 ): - await self.aclose() + await self._reset_for_reinitialize() # Reset the counter self.reinitialize_counter = 0 else: @@ -1196,9 +1211,7 @@ def get_node( return self.nodes_cache.get(node_name) else: raise DataError( - "get_node requires one of the following: " - "1. node name " - "2. host and port" + "get_node requires one of the following: 1. node name 2. host and port" ) def set_nodes( @@ -1374,7 +1387,7 @@ async def initialize(self) -> None: if len(disagreements) > 5: raise ValkeyClusterException( f"startup_nodes could not agree on a valid " - f'slots cache: {", ".join(disagreements)}' + f"slots cache: {', '.join(disagreements)}" ) # Validate if all slots are covered or if we should try next startup node @@ -1579,7 +1592,7 @@ async def execute( if type(e) in self.__class__.ERRORS_ALLOW_RETRY: # Try again with the new cluster setup. exception = e - await self._client.aclose() + await self._client._reset_for_reinitialize() await asyncio.sleep(0.25) else: # All other errors should be raised.