add rpushb and xadd

asyncio_redis/encoders.py

@@ -62,11 +62,15 @@ class StringEncoder(BaseEncoder):
 
     def encode_from_native(self, data):
         """ string to bytes """
-        return data.encode(self.encoding)
+        if isinstance(data, str):
+            return data.encode(self.encoding)
+        return data
 
     def decode_to_native(self, data):
         """ bytes to string """
-        return data.decode(self.encoding)
+        if isinstance(data, bytes):
+            return data.decode(self.encoding)
+        return data
 
 
 class UTF8Encoder(StringEncoder):

asyncio_redis/protocol.py

@@ -1500,6 +1500,27 @@ def sadd(self, tr, key: NativeType, members: ListOf(NativeType)) -> int:
             *map(self.encode_from_native, members),
         )
 
+    @_query_command
+    def xadd(
+            self, tr, stream: NativeType, fields: dict, id: NativeType='*',
+    ) -> NativeType:
+        """
+        Appends a new entry to a stream.
+
+        :param stream: The name of the stream.
+        :param id: The ID of the entry. Use '*' to auto-generate an ID.
+        :param fields: A list of field-value pairs.
+        :return: The ID of the added entry.
+        """
+        flattened_list = [item for sublist in fields.items() for item in sublist]
+        return self._query(
+            tr,
+            b"xadd",
+            self.encode_from_native(stream),
+            self.encode_from_native(id),
+            *map(self.encode_from_native, flattened_list),
+        )
+
     @_query_command
     def srem(self, tr, key: NativeType, members: ListOf(NativeType)) -> int:
         """Remove one or more members from a set
@@ -1646,6 +1667,17 @@ def rpush(self, tr, key: NativeType, values: ListOf(NativeType)) -> int:
             *map(self.encode_from_native, values),
         )
 
+    @_query_command
+    def rpushb(self, tr, key: NativeType, values: bytes) -> int:
+        """Append one or multiple values to a list
+        """
+        return self._query(
+            tr,
+            b"rpush",
+            self.encode_from_native(key),
+            values,
+        )
+
     @_query_command
     def rpushx(self, tr, key: NativeType, value: NativeType) -> int:
         """Append a value to a list, only if the list exists