Repository: redis/redis-py
File: redis/commands/core.py (Python)

Code before:
        pieces = []
        if block is not None:
            if not isinstance(block, int) or block < 0:
                raise DataError("XREAD block must be a non-negative integer")
            pieces.append(b"BLOCK")
            pieces.append(str(block))
        if count is not None:
            if not isinstance(count, int) or count < 1:
                raise DataError("XREAD count must be a positive integer")
            pieces.append(b"COUNT")
            pieces.append(str(count))
        if not isinstance(streams, dict) or len(streams) == 0:
            raise DataError("XREAD streams must be a non empty dict")
        pieces.append(b"STREAMS")
        keys, values = zip(*streams.items())
        pieces.extend(keys)
        pieces.extend(values)
        response = self.execute_command("XREAD", *pieces, keys=keys)

        if inspect.iscoroutine(response):
            # Async client - wrap in coroutine that awaits and records
            async def _record_and_return():
                actual_response = await response
                from redis.asyncio.observability.recorder import (
                    record_streaming_lag_from_response as async_record_streaming_lag,
                )

                await async_record_streaming_lag(response=actual_response)
                return actual_response

            return _record_and_return()
        else:
            # Sync client
            record_streaming_lag_from_response(response=response)
            return response

    def xreadgroup(
        self,
        groupname: str,
        consumername: str,
        streams: Dict[KeyT, StreamIdT],
        count: Optional[int] = None,
        block: Optional[int] = None,
        noack: bool = False,
        claim_min_idle_time: Optional[int] = None,
    ) -> ResponseT:
        """
        Read from a stream via a consumer group.

        groupname: name of the consumer group.


Review comment (performance):
Importing the async recorder inside the function could have performance implications since this function may be called frequently. Consider importing it at the module level similar to the sync version, or use a conditional import at the top of the file to avoid repeated imports on each function call.

Code after:
        For more information, see https://redis.io/commands/xread
        """
        pieces = []
        if block is not None:
            if not isinstance(block, int) or block < 0:
                raise DataError("XREAD block must be a non-negative integer")
            pieces.append(b"BLOCK")
            pieces.append(str(block))
        if count is not None:
            if not isinstance(count, int) or count < 1:
                raise DataError("XREAD count must be a positive integer")
            pieces.append(b"COUNT")
            pieces.append(str(count))
        if not isinstance(streams, dict) or len(streams) == 0:
            raise DataError("XREAD streams must be a non empty dict")
        pieces.append(b"STREAMS")
        keys, values = zip(*streams.items())
        pieces.extend(keys)
        pieces.extend(values)
        response = self.execute_command("XREAD", *pieces, keys=keys)

        if inspect.iscoroutine(response):
            # Async client - wrap in coroutine that awaits and records
            async def _record_and_return():
                actual_response = await response

                await async_record_streaming_lag(response=actual_response)
                return actual_response

            return _record_and_return()
        else:
            # Sync client
            record_streaming_lag_from_response(response=response)
            return response

    def xreadgroup(
        self,
        groupname: str,
        consumername: str,
        streams: Dict[KeyT, StreamIdT],
        count: Optional[int] = None,
        block: Optional[int] = None,
        noack: bool = False,
        claim_min_idle_time: Optional[int] = None,
    ) -> ResponseT:
        """
        Read from a stream via a consumer group.

        groupname: name of the consumer group.

        consumername: name of the requesting consumer.