Repository: redis/redis-py
File: redis/asyncio/client.py (Python)

Code before:

    async def execute(self, raise_on_error: bool = True):
        """Execute all the commands in the current pipeline"""
        stack = self.command_stack
        if not stack and not self.watching:
            return []
        if self.scripts:
            await self.load_scripts()
        if self.is_transaction or self.explicit_transaction:
            execute = self._execute_transaction
        else:
            execute = self._execute_pipeline

        conn = self.connection
        if not conn:
            conn = await self.connection_pool.get_connection("MULTI", self.shard_hint)
            # assign to self.connection so reset() releases the connection
            # back to the pool after we're done
            self.connection = conn
        conn = cast(Connection, conn)

        try:
            return await asyncio.shield(
                self._try_execute(conn, execute, stack, raise_on_error)
            )
        except RuntimeError:
            self.reset()

    async def discard(self):
        """Flushes all previously queued commands
        See: https://redis.io/commands/DISCARD
        """
        await self.execute_command("DISCARD")

    async def watch(self, *names: KeyT):
        """Watches the values at keys ``names``"""
        if self.explicit_transaction:
            raise RedisError("Cannot issue a WATCH after a MULTI")
        return await self.execute_command("WATCH", *names)

    async def unwatch(self):
        """Unwatches all previously specified keys"""
        return self.watching and await self.execute_command("UNWATCH") or True


Review comment (suggestion):
What happens if we Canceled (so this block raises CanceledError) and then get a RuntimeError inside the shield (unhandled by this try...except since we already exited this block)

Code after:

    async def execute(self, raise_on_error: bool = True):
        """Execute all the commands in the current pipeline"""
        stack = self.command_stack
        if not stack and not self.watching:
            return []
        if self.scripts:
            await self.load_scripts()
        if self.is_transaction or self.explicit_transaction:
            execute = self._execute_transaction
        else:
            execute = self._execute_pipeline

        conn = self.connection
        if not conn:
            conn = await self.connection_pool.get_connection("MULTI", self.shard_hint)
            # assign to self.connection so reset() releases the connection
            # back to the pool after we're done
            self.connection = conn
        conn = cast(Connection, conn)

        try:
            return await asyncio.shield(
                self._try_execute(conn, execute, stack, raise_on_error)
            )
        except RuntimeError:
            await self.reset()
        finally:
            await self.reset()

    async def discard(self):
        """Flushes all previously queued commands
        See: https://redis.io/commands/DISCARD
        """
        await self.execute_command("DISCARD")

    async def watch(self, *names: KeyT):
        """Watches the values at keys ``names``"""
        if self.explicit_transaction:
            raise RedisError("Cannot issue a WATCH after a MULTI")
        return await self.execute_command("WATCH", *names)

    async def unwatch(self):
        """Unwatches all previously specified keys"""
        return self.watching and await self.execute_command("UNWATCH") or True