use PEP 688 collections.abc.Buffer for type hints

Python 3.12 introduces a new type to indicate the C buffer protocol. We can use this to more correctly indicate allowable types.
jpstotz · Sep 2, 2023 · f3db1bb · f3db1bb
1 parent a8affc8
commit f3db1bb
Show file tree

Hide file tree

Showing 8 changed files with 54 additions and 36 deletions.
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
@@ -26,6 +26,7 @@ Changed
 * Added missing permissions and requirements in android kivy example. Fixes #1184.
 * Bleak recipe now automatically installs bleak from github release.
 * Changed `BlueZManager` methods to raise `BleakError` when device is not in BlueZ.
+* Changed type hint for buffer protocol to ``collections.abc.Buffer``.
 
 Fixed
 -----

diff --git a/bleak/__init__.py b/bleak/__init__.py
@@ -31,6 +31,11 @@
 )
 from warnings import warn
 
+if sys.version_info < (3, 12):
+    from typing_extensions import Buffer
+else:
+    from collections.abc import Buffer
+
 if sys.version_info < (3, 11):
     from async_timeout import timeout as async_timeout
 else:
@@ -671,7 +676,7 @@ async def read_gatt_char(
     async def write_gatt_char(
         self,
         char_specifier: Union[BleakGATTCharacteristic, int, str, uuid.UUID],
-        data: Union[bytes, bytearray, memoryview],
+        data: Buffer,
         response: bool = None,
     ) -> None:
         """
@@ -822,9 +827,7 @@ async def read_gatt_descriptor(self, handle: int, **kwargs) -> bytearray:
         """
         return await self._backend.read_gatt_descriptor(handle, **kwargs)
 
-    async def write_gatt_descriptor(
-        self, handle: int, data: Union[bytes, bytearray, memoryview]
-    ) -> None:
+    async def write_gatt_descriptor(self, handle: int, data: Buffer) -> None:
         """
         Perform a write operation on the specified GATT descriptor.
 

diff --git a/bleak/backends/bluezdbus/client.py b/bleak/backends/bluezdbus/client.py
@@ -10,6 +10,11 @@
 from typing import Callable, Dict, Optional, Set, Union, cast
 from uuid import UUID
 
+if sys.version_info < (3, 12):
+    from typing_extensions import Buffer
+else:
+    from collections.abc import Buffer
+
 if sys.version_info < (3, 11):
     from async_timeout import timeout as async_timeout
 else:
@@ -812,7 +817,7 @@ async def read_gatt_descriptor(self, handle: int, **kwargs) -> bytearray:
     async def write_gatt_char(
         self,
         characteristic: BleakGATTCharacteristic,
-        data: Union[bytes, bytearray, memoryview],
+        data: Buffer,
         response: bool,
     ) -> None:
         if not self.is_connected:
@@ -883,14 +888,12 @@ async def write_gatt_char(
             data,
         )
 
-    async def write_gatt_descriptor(
-        self, handle: int, data: Union[bytes, bytearray, memoryview]
-    ) -> None:
+    async def write_gatt_descriptor(self, handle: int, data: Buffer) -> None:
         """Perform a write operation on the specified GATT descriptor.
 
         Args:
-            handle (int): The handle of the descriptor to read from.
-            data (bytes or bytearray): The data to send.
+            handle: The handle of the descriptor to read from.
+            data: The data to send (any bytes-like object).
 
         """
         if not self.is_connected:

diff --git a/bleak/backends/client.py b/bleak/backends/client.py
@@ -9,10 +9,16 @@
 import asyncio
 import os
 import platform
+import sys
 import uuid
 from typing import Callable, Optional, Type, Union
 from warnings import warn
 
+if sys.version_info < (3, 12):
+    from typing_extensions import Buffer
+else:
+    from collections.abc import Buffer
+
 from ..exc import BleakError
 from .service import BleakGATTServiceCollection
 from .characteristic import BleakGATTCharacteristic
@@ -184,7 +190,7 @@ async def read_gatt_descriptor(self, handle: int, **kwargs) -> bytearray:
     async def write_gatt_char(
         self,
         characteristic: BleakGATTCharacteristic,
-        data: Union[bytes, bytearray, memoryview],
+        data: Buffer,
         response: bool,
     ) -> None:
         """
@@ -198,14 +204,12 @@ async def write_gatt_char(
         raise NotImplementedError()
 
     @abc.abstractmethod
-    async def write_gatt_descriptor(
-        self, handle: int, data: Union[bytes, bytearray, memoryview]
-    ) -> None:
+    async def write_gatt_descriptor(self, handle: int, data: Buffer) -> None:
         """Perform a write operation on the specified GATT descriptor.
 
         Args:
-            handle (int): The handle of the descriptor to read from.
-            data (bytes or bytearray): The data to send.
+            handle: The handle of the descriptor to read from.
+            data: The data to send (any bytes-like object).
 
         """
         raise NotImplementedError()

diff --git a/bleak/backends/corebluetooth/client.py b/bleak/backends/corebluetooth/client.py
@@ -5,9 +5,15 @@
 """
 import asyncio
 import logging
+import sys
 import uuid
 from typing import Optional, Set, Union
 
+if sys.version_info < (3, 12):
+    from typing_extensions import Buffer
+else:
+    from collections.abc import Buffer
+
 from CoreBluetooth import (
     CBUUID,
     CBCharacteristicWriteWithoutResponse,
@@ -308,7 +314,7 @@ async def read_gatt_descriptor(
     async def write_gatt_char(
         self,
         characteristic: BleakGATTCharacteristic,
-        data: Union[bytes, bytearray, memoryview],
+        data: Buffer,
         response: bool,
     ) -> None:
         value = NSData.alloc().initWithBytes_length_(data, len(data))
@@ -321,14 +327,12 @@ async def write_gatt_char(
         )
         logger.debug(f"Write Characteristic {characteristic.uuid} : {data}")
 
-    async def write_gatt_descriptor(
-        self, handle: int, data: Union[bytes, bytearray, memoryview]
-    ) -> None:
+    async def write_gatt_descriptor(self, handle: int, data: Buffer) -> None:
         """Perform a write operation on the specified GATT descriptor.
 
         Args:
-            handle (int): The handle of the descriptor to read from.
-            data (bytes or bytearray): The data to send.
+            handle: The handle of the descriptor to read from.
+            data: The data to send (any bytes-like object).
 
         """
         descriptor = self.services.get_descriptor(handle)

diff --git a/bleak/backends/winrt/client.py b/bleak/backends/winrt/client.py
@@ -13,6 +13,11 @@
 from ctypes import WinError
 from typing import Any, Dict, List, Optional, Sequence, Set, Union, cast
 
+if sys.version_info < (3, 12):
+    from typing_extensions import Buffer
+else:
+    from collections.abc import Buffer
+
 if sys.version_info < (3, 11):
     from async_timeout import timeout as async_timeout
 else:
@@ -53,7 +58,7 @@
     EventRegistrationToken,
     IAsyncOperation,
 )
-from bleak_winrt.windows.storage.streams import Buffer
+from bleak_winrt.windows.storage.streams import Buffer as WinBuffer
 
 from ... import BleakScanner
 from ...exc import PROTOCOL_ERROR_CODES, BleakDeviceNotFoundError, BleakError
@@ -836,7 +841,7 @@ async def read_gatt_descriptor(self, handle: int, **kwargs) -> bytearray:
     async def write_gatt_char(
         self,
         characteristic: BleakGATTCharacteristic,
-        data: Union[bytes, bytearray, memoryview],
+        data: Buffer,
         response: bool,
     ) -> None:
         if not self.is_connected:
@@ -847,7 +852,7 @@ async def write_gatt_char(
             if response
             else GattWriteOption.WRITE_WITHOUT_RESPONSE
         )
-        buf = Buffer(len(data))
+        buf = WinBuffer(len(data))
         buf.length = buf.capacity
         with memoryview(buf) as mv:
             mv[:] = data
@@ -857,14 +862,12 @@ async def write_gatt_char(
             f"Could not write value {data} to characteristic {characteristic.handle:04X}",
         )
 
-    async def write_gatt_descriptor(
-        self, handle: int, data: Union[bytes, bytearray, memoryview]
-    ) -> None:
+    async def write_gatt_descriptor(self, handle: int, data: Buffer) -> None:
         """Perform a write operation on the specified GATT descriptor.
 
         Args:
-            handle (int): The handle of the descriptor to read from.
-            data (bytes or bytearray): The data to send.
+            handle: The handle of the descriptor to read from.
+            data: The data to send (any bytes-like object).
 
         """
         if not self.is_connected:
@@ -874,7 +877,7 @@ async def write_gatt_descriptor(
         if not descriptor:
             raise BleakError(f"Descriptor with handle {handle} was not found!")
 
-        buf = Buffer(len(data))
+        buf = WinBuffer(len(data))
         buf.length = buf.capacity
         with memoryview(buf) as mv:
             mv[:] = data

diff --git a/poetry.lock b/poetry.lock
diff --git a/pyproject.toml b/pyproject.toml
@@ -24,7 +24,7 @@ classifiers = [
 [tool.poetry.dependencies]
 python = "^3.7"
 async-timeout = { version = ">= 3.0.0, < 5", python = "<3.11" }
-typing-extensions = { version = "^4.2.0", python = "<3.8" }
+typing-extensions = { version = ">=4.7.0", python = "<3.12" }
 pyobjc-core = { version = "^9.0.1", markers = "platform_system=='Darwin'" }
 pyobjc-framework-CoreBluetooth = { version = "^9.0.1", markers = "platform_system=='Darwin'" }
 pyobjc-framework-libdispatch = { version = "^9.0.1", markers = "platform_system=='Darwin'" }