use redis-py url syntax for redis_url (#2267)

* use redis-py url syntax for redis_url

* port is optional

* Add StateManagerRedis.close method

The close helper method always calls `close_connection_pool=True` so that all
outstanding redis operations can be stopped before changing event loops.

---------

Co-authored-by: Masen Furer <m_github@0x26.net>

.github/workflows/integration_app_harness.yml

@@ -45,7 +45,7 @@ jobs:
     - name: Run app harness tests
       env:
         SCREENSHOT_DIR: /tmp/screenshots
-        REDIS_URL: ${{ matrix.state_manager == 'redis' && 'localhost:6379' || '' }}
+        REDIS_URL: ${{ matrix.state_manager == 'redis' && 'redis://localhost:6379' || '' }}
       run: |
         poetry run pytest integration
     - uses: actions/upload-artifact@v3
@@ -53,4 +53,4 @@ jobs:
       if: always()
       with:
         name: failed_test_screenshots
-        path: /tmp/screenshots
+        path: /tmp/screenshots

.github/workflows/unit_tests.yml

@@ -70,6 +70,6 @@ jobs:
       if: ${{ matrix.os == 'ubuntu-latest' }}
       run: |
         export PYTHONUNBUFFERED=1
-        export REDIS_URL=localhost:6379
+        export REDIS_URL=redis://localhost:6379
         poetry run pytest tests --cov --no-cov-on-fail --cov-report=
     - run: poetry run coverage html

reflex/state.py

@@ -1788,6 +1788,17 @@ class StateManagerRedis(StateManager):
                 # only delete our lock
                 await self.redis.delete(lock_key)
 
+    async def close(self):
+        """Explicitly close the redis connection and connection_pool.
+
+        It is necessary in testing scenarios to close between asyncio test cases
+        to avoid having lingering redis connections associated with event loops
+        that will be closed (each test case uses its own event loop).
+
+        Note: Connections will be automatically reopened when needed.
+        """
+        await self.redis.close(close_connection_pool=True)
+
 
 class ClientStorageBase:
     """Base class for client-side storage."""

reflex/testing.py

@@ -184,7 +184,7 @@ class AppHarness:
             if self.app_instance is not None and isinstance(
                 self.app_instance.state_manager, StateManagerRedis
             ):
-                await self.app_instance.state_manager.redis.close()
+                await self.app_instance.state_manager.close()
             await original_shutdown(*args, **kwargs)
 
         return _shutdown_redis
@@ -455,7 +455,7 @@ class AppHarness:
             return await self.state_manager.get_state(token)
         finally:
             if isinstance(self.state_manager, StateManagerRedis):
-                await self.state_manager.redis.close()
+                await self.state_manager.close()
 
     async def set_state(self, token: str, **kwargs) -> None:
         """Set the state associated with the given token.
@@ -476,7 +476,7 @@ class AppHarness:
             await self.state_manager.set_state(token, state)
         finally:
             if isinstance(self.state_manager, StateManagerRedis):
-                await self.state_manager.redis.close()
+                await self.state_manager.close()
 
     @contextlib.asynccontextmanager
     async def modify_state(self, token: str) -> AsyncIterator[State]:
@@ -506,7 +506,7 @@ class AppHarness:
         finally:
             if isinstance(self.state_manager, StateManagerRedis):
                 self.app_instance._state_manager = app_state_manager
-                await self.state_manager.redis.close()
+                await self.state_manager.close()
 
     def poll_for_content(
         self,

reflex/utils/prerequisites.py

@@ -173,6 +173,14 @@ def get_redis() -> Redis | None:
     config = get_config()
     if not config.redis_url:
         return None
+    if config.redis_url.startswith(("redis://", "rediss://", "unix://")):
+        return Redis.from_url(config.redis_url)
+    console.deprecate(
+        feature_name="host[:port] style redis urls",
+        reason="redis-py url syntax is now being used",
+        deprecation_version="0.3.6",
+        removal_version="0.4.0",
+    )
     redis_url, has_port, redis_port = config.redis_url.partition(":")
     if not has_port:
         redis_port = 6379

tests/test_app.py

@@ -344,7 +344,7 @@ async def test_initialize_with_state(test_state: Type[ATestState], token: str):
     assert state.var == 0  # type: ignore
 
     if isinstance(app.state_manager, StateManagerRedis):
-        await app.state_manager.redis.close()
+        await app.state_manager.close()
 
 
 @pytest.mark.asyncio
@@ -379,7 +379,7 @@ async def test_set_and_get_state(test_state):
     assert state2.var == 2  # type: ignore
 
     if isinstance(app.state_manager, StateManagerRedis):
-        await app.state_manager.redis.close()
+        await app.state_manager.close()
 
 
 @pytest.mark.asyncio
@@ -781,7 +781,7 @@ async def test_upload_file(tmp_path, state, delta, token: str, mocker):
     ]
 
     if isinstance(app.state_manager, StateManagerRedis):
-        await app.state_manager.redis.close()
+        await app.state_manager.close()
 
 
 @pytest.mark.asyncio
@@ -817,7 +817,7 @@ async def test_upload_file_without_annotation(state, tmp_path, token):
     )
 
     if isinstance(app.state_manager, StateManagerRedis):
-        await app.state_manager.redis.close()
+        await app.state_manager.close()
 
 
 @pytest.mark.asyncio
@@ -853,7 +853,7 @@ async def test_upload_file_background(state, tmp_path, token):
     )
 
     if isinstance(app.state_manager, StateManagerRedis):
-        await app.state_manager.redis.close()
+        await app.state_manager.close()
 
 
 class DynamicState(BaseState):
@@ -1093,7 +1093,7 @@ async def test_dynamic_route_var_route_change_completed_on_load(
     # assert state.side_effect_counter == len(exp_vals)
 
     if isinstance(app.state_manager, StateManagerRedis):
-        await app.state_manager.redis.close()
+        await app.state_manager.close()
 
 
 @pytest.mark.asyncio
@@ -1127,7 +1127,7 @@ async def test_process_events(mocker, token: str):
     assert app.postprocess.call_count == 6
 
     if isinstance(app.state_manager, StateManagerRedis):
-        await app.state_manager.redis.close()
+        await app.state_manager.close()
 
 
 @pytest.mark.parametrize(

tests/test_state.py

@@ -1429,7 +1429,7 @@ def state_manager(request) -> Generator[StateManager, None, None]:
     yield state_manager
 
     if isinstance(state_manager, StateManagerRedis):
-        asyncio.get_event_loop().run_until_complete(state_manager.redis.close())
+        asyncio.get_event_loop().run_until_complete(state_manager.close())
 
 
 @pytest.mark.asyncio
@@ -1507,7 +1507,7 @@ def state_manager_redis() -> Generator[StateManager, None, None]:
 
     yield state_manager
 
-    asyncio.get_event_loop().run_until_complete(state_manager.redis.close())
+    asyncio.get_event_loop().run_until_complete(state_manager.close())
 
 
 @pytest.mark.asyncio