reformatting applied

Author: maxwellflitton

src/surrealdb/cbor2/_decoder.py

@@ -30,7 +30,8 @@
 T = TypeVar("T")
 
 timestamp_re = re.compile(
-    r"^(\d{4})-(\d\d)-(\d\d)T(\d\d):(\d\d):(\d\d)" r"(?:\.(\d{1,6})\d*)?(?:Z|([+-])(\d\d):(\d\d))$"
+    r"^(\d{4})-(\d\d)-(\d\d)T(\d\d):(\d\d):(\d\d)"
+    r"(?:\.(\d{1,6})\d*)?(?:Z|([+-])(\d\d):(\d\d))$"
 )
 incremental_utf8_decoder = getincrementaldecoder("utf-8")
 
@@ -141,7 +142,9 @@ def object_hook(self) -> Callable[[CBORDecoder, dict[Any, Any]], Any] | None:
         return self._object_hook
 
     @object_hook.setter
-    def object_hook(self, value: Callable[[CBORDecoder, Mapping[Any, Any]], Any] | None) -> None:
+    def object_hook(
+        self, value: Callable[[CBORDecoder, Mapping[Any, Any]], Any] | None
+    ) -> None:
         if value is None or callable(value):
             self._object_hook = value
         else:
@@ -253,9 +256,13 @@ def decode_from_bytes(self, buf: bytes) -> object:
     def _decode_length(self, subtype: int) -> int: ...
 
     @overload
-    def _decode_length(self, subtype: int, allow_indefinite: Literal[True]) -> int | None: ...
+    def _decode_length(
+        self, subtype: int, allow_indefinite: Literal[True]
+    ) -> int | None: ...
 
-    def _decode_length(self, subtype: int, allow_indefinite: bool = False) -> int | None:
+    def _decode_length(
+        self, subtype: int, allow_indefinite: bool = False
+    ) -> int | None:
         if subtype < 24:
             return subtype
         elif subtype == 24:
@@ -269,7 +276,9 @@ def _decode_length(self, subtype: int, allow_indefinite: bool = False) -> int |
         elif subtype == 31 and allow_indefinite:
             return None
         else:
-            raise CBORDecodeValueError(f"unknown unsigned integer subtype 0x{subtype:x}")
+            raise CBORDecodeValueError(
+                f"unknown unsigned integer subtype 0x{subtype:x}"
+            )
 
     def decode_uint(self, subtype: int) -> int:
         # Major tag 0
@@ -304,7 +313,9 @@ def decode_bytestring(self, subtype: int) -> bytes:
                     )
         else:
             if length > sys.maxsize:
-                raise CBORDecodeValueError(f"invalid length for bytestring 0x{length:x}")
+                raise CBORDecodeValueError(
+                    f"invalid length for bytestring 0x{length:x}"
+                )
             elif length <= 65536:
                 result = self.read(length)
             else:
@@ -356,11 +367,15 @@ def decode_string(self, subtype: int) -> str:
                     try:
                         value = self.read(length).decode("utf-8", self._str_errors)
                     except UnicodeDecodeError as exc:
-                        raise CBORDecodeValueError("error decoding unicode string") from exc
+                        raise CBORDecodeValueError(
+                            "error decoding unicode string"
+                        ) from exc
 
                     buf.append(value)
                 else:
-                    raise CBORDecodeValueError("non-string found in indefinite length string")
+                    raise CBORDecodeValueError(
+                        "non-string found in indefinite length string"
+                    )
         else:
             if length > sys.maxsize:
                 raise CBORDecodeValueError(f"invalid length for string 0x{length:x}")
@@ -381,7 +396,9 @@ def decode_string(self, subtype: int) -> str:
                     try:
                         result += codec.decode(self.read(chunk_size), final)
                     except UnicodeDecodeError as exc:
-                        raise CBORDecodeValueError("error decoding unicode string") from exc
+                        raise CBORDecodeValueError(
+                            "error decoding unicode string"
+                        ) from exc
 
                     left -= chunk_size
 
@@ -619,7 +636,9 @@ def decode_sharedref(self) -> Any:
             raise CBORDecodeValueError("shared reference %d not found" % value)
 
         if shared is None:
-            raise CBORDecodeValueError("shared value %d has not been initialized" % value)
+            raise CBORDecodeValueError(
+                "shared value %d has not been initialized" % value
+            )
         else:
             return shared
 

src/surrealdb/cbor2/_encoder.py

@@ -179,15 +179,19 @@ def __init__(
         self.string_namespacing = string_referencing
         self.default = default
         self._canonical = canonical
-        self._shared_containers: dict[
-            int, tuple[object, int | None]
-        ] = {}  # indexes used for value sharing
-        self._string_references: dict[str | bytes, int] = {}  # indexes used for string references
+        self._shared_containers: dict[int, tuple[object, int | None]] = (
+            {}
+        )  # indexes used for value sharing
+        self._string_references: dict[str | bytes, int] = (
+            {}
+        )  # indexes used for string references
         self._encoders = default_encoders.copy()
         if canonical:
             self._encoders.update(canonical_encoders)
 
-    def _find_encoder(self, obj_type: type) -> Callable[[CBOREncoder, Any], None] | None:
+    def _find_encoder(
+        self, obj_type: type
+    ) -> Callable[[CBOREncoder, Any], None] | None:
         for type_or_tuple, enc in list(self._encoders.items()):
             if type(type_or_tuple) is tuple:
                 try:
@@ -306,7 +310,11 @@ def encode(self, obj: Any) -> None:
             the object to encode
         """
         obj_type = obj.__class__
-        encoder = self._encoders.get(obj_type) or self._find_encoder(obj_type) or self._default
+        encoder = (
+            self._encoders.get(obj_type)
+            or self._find_encoder(obj_type)
+            or self._default
+        )
         if not encoder:
             raise CBOREncodeTypeError(f"cannot serialize type {obj_type.__name__}")
 
@@ -327,15 +335,19 @@ def encode_to_bytes(self, obj: Any) -> bytes:
             self.fp = old_fp
             return fp.getvalue()
 
-    def encode_container(self, encoder: Callable[[CBOREncoder, Any], Any], value: Any) -> None:
+    def encode_container(
+        self, encoder: Callable[[CBOREncoder, Any], Any], value: Any
+    ) -> None:
         if self.string_namespacing:
             # Create a new string reference domain
             self.encode_length(6, 256)
 
         with self.disable_string_namespacing():
             self.encode_shared(encoder, value)
 
-    def encode_shared(self, encoder: Callable[[CBOREncoder, Any], Any], value: Any) -> None:
+    def encode_shared(
+        self, encoder: Callable[[CBOREncoder, Any], Any], value: Any
+    ) -> None:
         value_id = id(value)
         try:
             index = self._shared_containers[id(value)][1]
@@ -470,7 +482,9 @@ def encode_sortable_key(self, value: Any) -> tuple[int, bytes]:
     @container_encoder
     def encode_canonical_map(self, value: Mapping[Any, Any]) -> None:
         """Reorder keys according to Canonical CBOR specification"""
-        keyed_keys = ((self.encode_sortable_key(key), key, value) for key, value in value.items())
+        keyed_keys = (
+            (self.encode_sortable_key(key), key, value) for key, value in value.items()
+        )
         self.encode_length(5, len(value))
         for sortkey, realkey, value in sorted(keyed_keys):
             if self.string_referencing:
@@ -507,7 +521,8 @@ def encode_datetime(self, value: datetime) -> None:
                 value = value.replace(tzinfo=self._timezone)
             else:
                 raise CBOREncodeValueError(
-                    f"naive datetime {value!r} encountered and no default timezone " "has been set"
+                    f"naive datetime {value!r} encountered and no default timezone "
+                    "has been set"
                 )
 
         if self.datetime_as_timestamp:
@@ -593,7 +608,9 @@ def encode_ipaddress(self, value: IPv4Address | IPv6Address) -> None:
 
     def encode_ipnetwork(self, value: IPv4Network | IPv6Network) -> None:
         # Semantic tag 261
-        self.encode_semantic(CBORTag(261, {value.network_address.packed: value.prefixlen}))
+        self.encode_semantic(
+            CBORTag(261, {value.network_address.packed: value.prefixlen})
+        )
 
     #
     # Special encoders (major tag 7)

src/surrealdb/cbor2/decoder.py

@@ -4,4 +4,6 @@
 from surrealdb.cbor2._decoder import load as load
 from surrealdb.cbor2._decoder import loads as loads
 
-warn("The cbor.decoder module has been deprecated. Instead import everything directly from cbor2.")
+warn(
+    "The cbor.decoder module has been deprecated. Instead import everything directly from cbor2."
+)

src/surrealdb/cbor2/tool.py

@@ -23,7 +23,9 @@
     from typing import Literal, TypeAlias
 
 T = TypeVar("T")
-JSONValue: TypeAlias = "str | float | bool | None | list[JSONValue] | dict[str, JSONValue]"
+JSONValue: TypeAlias = (
+    "str | float | bool | None | list[JSONValue] | dict[str, JSONValue]"
+)
 
 default_encoders: dict[type, Callable[[Any], Any]] = {
     bytes: lambda x: x.decode(encoding="utf-8", errors="backslashreplace"),
@@ -44,7 +46,9 @@
 }
 
 
-def tag_hook(decoder: CBORDecoder, tag: CBORTag, ignore_tags: Collection[int] = ()) -> object:
+def tag_hook(
+    decoder: CBORDecoder, tag: CBORTag, ignore_tags: Collection[int] = ()
+) -> object:
     if tag.tag in ignore_tags:
         return tag.value
 
@@ -72,15 +76,19 @@ def iterdecode(
     object_hook: Callable[[CBORDecoder, dict[Any, Any]], Any] | None = None,
     str_errors: Literal["strict", "error", "replace"] = "strict",
 ) -> Iterator[Any]:
-    decoder = CBORDecoder(f, tag_hook=tag_hook, object_hook=object_hook, str_errors=str_errors)
+    decoder = CBORDecoder(
+        f, tag_hook=tag_hook, object_hook=object_hook, str_errors=str_errors
+    )
     while True:
         try:
             yield decoder.decode()
         except EOFError:
             return
 
 
-def key_to_str(d: T, dict_ids: set[int] | None = None) -> str | list[Any] | dict[str, Any] | T:
+def key_to_str(
+    d: T, dict_ids: set[int] | None = None
+) -> str | list[Any] | dict[str, Any] | T:
     dict_ids = set(dict_ids or [])
     rval: dict[str, Any] = {}
     if not isinstance(d, dict):

src/surrealdb/cbor2/types.py

@@ -12,4 +12,6 @@
 from surrealdb.cbor2._types import FrozenDict as FrozenDict
 from surrealdb.cbor2._types import undefined as undefined
 
-warn("The cbor2.types module has been deprecated. Instead import everything directly from cbor2.")
+warn(
+    "The cbor2.types module has been deprecated. Instead import everything directly from cbor2."
+)

src/surrealdb/data/cbor.py

@@ -38,14 +38,10 @@ def default_encoder(encoder, obj):
         tagged = CBORTag(constants.TAG_GEOMETRY_MULTI_LINE, obj.get_coordinates())
 
     elif isinstance(obj, GeometryMultiPoint):
-        tagged = CBORTag(
-            constants.TAG_GEOMETRY_MULTI_POINT, obj.get_coordinates()
-        )
+        tagged = CBORTag(constants.TAG_GEOMETRY_MULTI_POINT, obj.get_coordinates())
 
     elif isinstance(obj, GeometryMultiPolygon):
-        tagged = CBORTag(
-            constants.TAG_GEOMETRY_MULTI_POLYGON, obj.get_coordinates()
-        )
+        tagged = CBORTag(constants.TAG_GEOMETRY_MULTI_POLYGON, obj.get_coordinates())
 
     elif isinstance(obj, GeometryCollection):
         tagged = CBORTag(constants.TAG_GEOMETRY_COLLECTION, obj.geometries)