Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/redis/asyncio/utils.py: 50%

1from typing import TYPE_CHECKING 

2 

3if TYPE_CHECKING: 

4 from redis.asyncio.client import Pipeline, Redis 

5 

6 

7def from_url(url, **kwargs): 

8 """ 

9 Returns an active Redis client generated from the given database URL. 

10 

11 Will attempt to extract the database id from the path url fragment, if 

12 none is provided. 

13 """ 

14 from redis.asyncio.client import Redis 

15 

16 return Redis.from_url(url, **kwargs) 

17 

18 

19class pipeline: # noqa: N801 

20 def __init__(self, redis_obj: "Redis"): 

21 self.p: "Pipeline" = redis_obj.pipeline() 

22 

23 async def __aenter__(self) -> "Pipeline": 

24 return self.p 

25 

26 async def __aexit__(self, exc_type, exc_value, traceback): 

27 await self.p.execute() 

28 del self.p