Coverage for faststream / redis / configs / state.py: 86%

1from typing import Any (empty)

2 

3from redis.asyncio.client import Redis (empty)

4from redis.asyncio.connection import ConnectionPool (empty)

5 

6from faststream.__about__ import __version__ (empty)

7from faststream.exceptions import IncorrectState (empty)

8 

9 

10class ConnectionState: (empty)

11 def __init__(self, options: dict[str, Any] | None = None) -> None: (empty)

12 self._options = options or {} (empty)

13 

14 self._connected = False (empty)

15 self._client: Redis[bytes] | None = None (empty)

16 

17 @property (empty)

18 def client(self) -> "Redis[bytes]": (empty)

19 if not self._client: 19 ↛ 20line 19 didn't jump to line 20 because the condition on line 19 was never true(empty)

20 msg = "Connection is not available yet. Please, connect the broker first." 

21 raise IncorrectState(msg) 

22 

23 return self._client (empty)

24 

25 def __bool__(self) -> bool: (empty)

26 return self._connected 

27 

28 async def connect(self) -> "Redis[bytes]": (empty)

29 pool = ConnectionPool( (empty)

30 **self._options, 

31 lib_name="faststream", 

32 lib_version=__version__, 

33 ) 

34 client: Redis[bytes] = Redis.from_pool(pool) # type: ignore[attr-defined] (empty)

35 

36 self._client = client (empty)

37 self._connected = True (empty)

38 

39 return client (empty)

40 

41 async def disconnect(self) -> None: (empty)

42 if self._client: (empty)

43 await self._client.aclose() # type: ignore[attr-defined] (empty)

44 

45 self._client = None (empty)

46 self._connected = False (empty)