PYTHON-6274 Enable ruff rule PLR2004

.evergreen/scripts/mod_wsgi_matrix.py

@@ -19,8 +19,10 @@
     },
 ]
 
+_EXPECTED_ARGC = 3
+
 if __name__ == "__main__":
-    if len(sys.argv) == 3 and sys.argv[1] == "--index":
+    if len(sys.argv) == _EXPECTED_ARGC and sys.argv[1] == "--index":
         # Emit one version pair in GITHUB_OUTPUT format.
         entry = VERSIONS[int(sys.argv[2])]
         print(f"python-version={entry['python-version']}")

.evergreen/scripts/setup_tests.py

@@ -74,6 +74,8 @@
 # (MONGOCRYPT-938).
 LIBMONGOCRYPT_VERSION = "1.21.0"
 
+_HTTP_OK = 200
+
 
 def is_set(var: str) -> bool:
     value = os.environ.get(var, "")
@@ -149,7 +151,7 @@ def setup_libmongocrypt():
 
     LOGGER.info(f"Fetching {url}...")
     with request.urlopen(request.Request(url), timeout=15.0) as response:  # noqa: S310
-        if response.status == 200:
+        if response.status == _HTTP_OK:
             with Path("libmongocrypt.tar.gz").open("wb") as f:
                 f.write(response.read())
             Path("libmongocrypt").mkdir()

.evergreen/scripts/test_generate_config.py

@@ -16,6 +16,10 @@
 import generate_config
 import generate_config_utils
 
+_MIN_PR_TASKS = 2
+_MAX_STANDARD_PR_TASKS = 5
+_MAX_NEWER_STANDARD_PR_TASKS = 3
+
 
 def all_tasks():
     tasks = []
@@ -160,7 +164,7 @@ def test_pr_tasks_cover_version_boundaries():
         assert seen == set(pythons), (
             f"{version} pr tasks pin {sorted(seen)}, want {sorted(pythons)}"
         )
-        assert len(tasks) >= 2, f"{version} has too few server-version pr tasks"
+        assert len(tasks) >= _MIN_PR_TASKS, f"{version} has too few server-version pr tasks"
 
 
 def test_server_version_rotated_tasks_keep_coverage():
@@ -203,7 +207,9 @@ def test_standard_family_pr_budget():
     filter).
     """
     tasks = [task for task in pr_tasks() if "test-standard" in (task.tags or [])]
-    assert len(tasks) <= 5, f"too many test-standard PR tasks: {[t.name for t in tasks]}"
+    assert len(tasks) <= _MAX_STANDARD_PR_TASKS, (
+        f"too many test-standard PR tasks: {[t.name for t in tasks]}"
+    )
     newer = [
         task
         for task in tasks
@@ -212,7 +218,9 @@ def test_standard_family_pr_budget():
             for tag in ("python-3.14", "python-3.14t", "python-3.15", "python-3.15t")
         )
     ]
-    assert len(newer) <= 3, f"too many 3.14+ test-standard PR tasks: {[t.name for t in newer]}"
+    assert len(newer) <= _MAX_NEWER_STANDARD_PR_TASKS, (
+        f"too many 3.14+ test-standard PR tasks: {[t.name for t in newer]}"
+    )
 
 
 def test_pr_tasks_run_coverage():

bson/__init__.py

@@ -98,6 +98,7 @@
     ALL_UUID_SUBTYPES,
     CSHARP_LEGACY,
     JAVA_LEGACY,
+    OLD_BINARY_SUBTYPE,
     OLD_UUID_SUBTYPE,
     STANDARD,
     UUID_SUBTYPE,
@@ -228,6 +229,12 @@
 BSONMIN = b"\xff"  # Min key
 BSONMAX = b"\x7f"  # Max key
 
+_INT32_MIN = -(2**31)
+_INT32_MAX = 2**31 - 1
+_INT32_SIZE = 4
+# A BSON document is at minimum the 4-byte length prefix and the EOO byte.
+_MIN_DOCUMENT_SIZE = 5
+
 
 _UNPACK_FLOAT_FROM = struct.Struct("<d").unpack_from
 _UNPACK_INT = struct.Struct("<i").unpack
@@ -392,7 +399,7 @@ def _get_binary(
     """Decode a BSON binary to bson.binary.Binary or python UUID."""
     length, subtype = _UNPACK_LENGTH_SUBTYPE_FROM(data, position)
     position += 5
-    if subtype == 2:
+    if subtype == OLD_BINARY_SUBTYPE:
         length2 = _UNPACK_INT_FROM(data, position)[0]
         position += 4
         if length2 != length - 4:
@@ -783,7 +790,7 @@ def _encode_text(name: bytes, value: str, dummy0: Any, dummy1: Any) -> bytes:
 def _encode_binary(name: bytes, value: Binary, dummy0: Any, dummy1: Any) -> bytes:
     """Encode bson.binary.Binary."""
     subtype = value.subtype
-    if subtype == 2:
+    if subtype == OLD_BINARY_SUBTYPE:
         value = _PACK_INT(len(value)) + value  # type: ignore
     return b"\x05" + name + _PACK_LENGTH_SUBTYPE(len(value), subtype) + value
 
@@ -861,7 +868,7 @@ def _encode_code(name: bytes, value: Code, dummy: Any, opts: CodecOptions[Any])
 
 def _encode_int(name: bytes, value: int, dummy0: Any, dummy1: Any) -> bytes:
     """Encode a python int."""
-    if -2147483648 <= value <= 2147483647:
+    if _INT32_MIN <= value <= _INT32_MAX:
         return b"\x10" + name + _PACK_INT(value)
     else:
         try:
@@ -1134,10 +1141,10 @@ def _decode_all(data: _ReadableBuffer, opts: CodecOptions[_DocumentType]) -> lis
     try:
         while position < data_len:
             # A document is at minimum 5 bytes: the 4-byte size and the EOO.
-            if data_len - position < 5:
+            if data_len - position < _MIN_DOCUMENT_SIZE:
                 raise InvalidBSON("not enough data for a BSON document")
             obj_size = _UNPACK_INT_FROM(data, position)[0]
-            if obj_size < 5:
+            if obj_size < _MIN_DOCUMENT_SIZE:
                 raise InvalidBSON("invalid message size")
             if data_len - position < obj_size:
                 raise InvalidBSON("objsize too large")
@@ -1397,9 +1404,9 @@ def decode_file_iter(
         size_data: Any = file_obj.read(4)
         if not size_data:
             break  # Finished with file normally.
-        elif len(size_data) != 4:
+        elif len(size_data) != _INT32_SIZE:
             raise InvalidBSON("cut off in middle of objsize")
-        obj_size = _UNPACK_INT_FROM(size_data, 0)[0] - 4
+        obj_size = _UNPACK_INT_FROM(size_data, 0)[0] - _INT32_SIZE
         elements = size_data + file_obj.read(max(0, obj_size))
         yield _bson_to_dict(elements, opts)  # type:ignore[misc]
 

bson/binary.py

@@ -209,6 +209,15 @@ class UuidRepresentation:
 """BSON binary subtype for any user defined structure.
 """
 
+# Binary subtypes are one byte, so the valid range is [0, 256).
+_SUBTYPE_LIMIT = 256
+_UUID_LENGTH = 16
+_MAX_PADDING = 7
+_INT8_MIN = -128
+_INT8_MAX = 127
+# Packed-bit vectors are restricted to the signed int8 range when built from numpy arrays.
+_PACKED_BIT_MAX = 127
+
 
 class BinaryVectorDtype(Enum):
     """Datatypes of vector subtype.
@@ -316,7 +325,7 @@ def __new__(
     ) -> Self:
         if not isinstance(subtype, int):
             raise TypeError(f"subtype must be an instance of int, not {type(subtype)}")
-        if subtype >= 256 or subtype < 0:
+        if subtype >= _SUBTYPE_LIMIT or subtype < 0:
             raise ValueError("subtype must be contained in [0, 256)")
         # Support any type that implements the buffer protocol.
         self = bytes.__new__(cls, memoryview(data).tobytes())
@@ -405,8 +414,10 @@ def as_uuid(self, uuid_representation: int = UuidRepresentation.STANDARD) -> UUI
         if uuid_representation == UuidRepresentation.UNSPECIFIED:
             raise ValueError("uuid_representation cannot be UNSPECIFIED")
 
-        if len(self) != 16:
-            raise ValueError(f"cannot decode a uuid from a {len(self)}-byte payload, expected 16")
+        if len(self) != _UUID_LENGTH:
+            raise ValueError(
+                f"cannot decode a uuid from a {len(self)}-byte payload, expected {_UUID_LENGTH}"
+            )
 
         if uuid_representation == UuidRepresentation.PYTHON_LEGACY:
             if self.subtype == OLD_UUID_SUBTYPE:
@@ -496,7 +507,7 @@ def from_vector(
                     raise ValueError(f"padding does not apply to {dtype=}")
             elif dtype == BinaryVectorDtype.PACKED_BIT:  # pack ints in [0, 255] as unsigned uint8
                 format_str = "B"
-                if 0 <= padding > 7:
+                if 0 <= padding > _MAX_PADDING:
                     raise ValueError(f"{padding=}. It must be in [0,1, ..7].")
                 if padding and not vector:
                     raise ValueError("Empty vector with non-zero padding.")
@@ -526,12 +537,12 @@ def from_vector(
             if dtype == BinaryVectorDtype.FLOAT32:
                 vector = vector.astype(np.dtype("float32"), copy=False)
             elif dtype == BinaryVectorDtype.INT8:
-                if vector.min() >= -128 and vector.max() <= 127:
+                if vector.min() >= _INT8_MIN and vector.max() <= _INT8_MAX:
                     vector = vector.astype(np.dtype("int8"), copy=False)
                 else:
                     raise ValueError("Values found outside INT8 range.")
             elif dtype == BinaryVectorDtype.PACKED_BIT:
-                if vector.min() >= 0 and vector.max() <= 127:
+                if vector.min() >= 0 and vector.max() <= _PACKED_BIT_MAX:
                     vector = vector.astype(np.dtype("uint8"), copy=False)
                 else:
                     raise ValueError("Values found outside UINT8 range.")
@@ -588,8 +599,10 @@ def as_vector(self, return_numpy: bool = False) -> BinaryVector:
                 # data packed as uint8
                 if padding and not n_bytes:
                     raise ValueError("Corrupt data. Vector has a padding P, but no data.")
-                if padding > 7 or padding < 0:
-                    raise ValueError(f"Corrupt data. Padding ({padding}) must be between 0 and 7.")
+                if padding > _MAX_PADDING or padding < 0:
+                    raise ValueError(
+                        f"Corrupt data. Padding ({padding}) must be between 0 and {_MAX_PADDING}."
+                    )
                 dtype_format = "B"
                 format_string = f"<{n_bytes}{dtype_format}"
                 unpacked_uint8s = list(struct.unpack_from(format_string, self, offset))
@@ -622,8 +635,10 @@ def as_vector(self, return_numpy: bool = False) -> BinaryVector:
                 # data packed as uint8
                 if padding and not n_bytes:
                     raise ValueError("Corrupt data. Vector has a padding P, but no data.")
-                if padding > 7 or padding < 0:
-                    raise ValueError(f"Corrupt data. Padding ({padding}) must be between 0 and 7.")
+                if padding > _MAX_PADDING or padding < 0:
+                    raise ValueError(
+                        f"Corrupt data. Padding ({padding}) must be between 0 and {_MAX_PADDING}."
+                    )
                 data = np.frombuffer(self[offset:], dtype="uint8")
                 if padding and np.unpackbits(data[-1])[-padding:].sum() > 0:
                     warnings.warn(

bson/decimal128.py

@@ -36,6 +36,7 @@
 _EXPONENT_MIN = -6143
 _MAX_DIGITS = 34
 _MAX_COEFFICIENT = 10**_MAX_DIGITS - 1
+_BYTE_LENGTH = 16
 
 _INF = 0x7800000000000000
 _NAN = 0x7C00000000000000
@@ -259,7 +260,8 @@ def __init__(self, value: _VALUE_OPTIONS) -> None:
         if isinstance(value, (str, decimal.Decimal)):
             self.__high, self.__low = _decimal_to_128(value)
         elif isinstance(value, (list, tuple)):
-            if len(value) != 2:
+            # A literal length is required for mypy to narrow the tuple type.
+            if len(value) != 2:  # noqa: PLR2004
                 raise ValueError(
                     "Invalid size for creation of Decimal128 "
                     "from list or tuple. Must have exactly 2 "
@@ -325,7 +327,7 @@ def from_bid(cls: type[Decimal128], value: bytes) -> Decimal128:
         """
         if not isinstance(value, bytes):
             raise TypeError(f"value must be an instance of bytes, not {type(value)}")
-        if len(value) != 16:
+        if len(value) != _BYTE_LENGTH:
             raise ValueError("value must be exactly 16 bytes")
         return cls((_UNPACK_64(value[8:])[0], _UNPACK_64(value[:8])[0]))  # type: ignore
 

bson/json_util.py

@@ -140,6 +140,17 @@
 from bson.timestamp import Timestamp
 from bson.tz_util import utc
 
+# Legacy mongoexport subtype values use the top 32 bits for a negative sign.
+_MONGOEXPORT_SUBTYPE_MASK = 0xFFFFFF80
+_MAX_SUBTYPE_HEX_LENGTH = 2
+_BINARY_COMPONENT_COUNT = 2
+_REGEX_COMPONENT_COUNT = 2
+_DBREF_COMPONENT_COUNT = 2
+_SINGLE_FIELD_COUNT = 1
+_TZ_OFFSET_HH_COLON_MM_LENGTH = 6
+_TZ_OFFSET_HHMM_LENGTH = 5
+_TZ_OFFSET_HH_LENGTH = 3
+
 _RE_OPT_TABLE = {
     "i": re.IGNORECASE,
     "l": re.LOCALE,
@@ -565,7 +576,7 @@ def _parse_legacy_binary(doc: Any, json_options: JSONOptions) -> Binary | uuid.U
     if isinstance(doc["$type"], int):
         doc["$type"] = "{:02x}".format(doc["$type"])
     subtype = int(doc["$type"], 16)
-    if subtype >= 0xFFFFFF80:  # Handle mongoexport values
+    if subtype >= _MONGOEXPORT_SUBTYPE_MASK:  # Handle mongoexport values
         subtype = int(doc["$type"][6:], 16)
     data = base64.b64decode(doc["$binary"].encode())
     return _binary_or_uuid(data, subtype, json_options)
@@ -577,9 +588,9 @@ def _parse_canonical_binary(doc: Any, json_options: JSONOptions) -> Binary | uui
     subtype = binary["subType"]
     if not isinstance(b64, str):
         raise TypeError(f"$binary base64 must be a string: {doc}")
-    if not isinstance(subtype, str) or len(subtype) > 2:
+    if not isinstance(subtype, str) or len(subtype) > _MAX_SUBTYPE_HEX_LENGTH:
         raise TypeError(f"$binary subType must be a string at most 2 characters: {doc}")
-    if len(binary) != 2:
+    if len(binary) != _BINARY_COMPONENT_COUNT:
         raise TypeError(f'$binary must include only "base64" and "subType" components: {doc}')
 
     data = base64.b64decode(b64.encode())
@@ -630,12 +641,12 @@ def _parse_canonical_datetime(
         )
 
         if offset and offset != "Z":
-            if len(offset) == 6:
+            if len(offset) == _TZ_OFFSET_HH_COLON_MM_LENGTH:
                 hours, minutes = offset[1:].split(":")
                 secs = int(hours) * 3600 + int(minutes) * 60
-            elif len(offset) == 5:
+            elif len(offset) == _TZ_OFFSET_HHMM_LENGTH:
                 secs = int(offset[1:3]) * 3600 + int(offset[3:]) * 60
-            elif len(offset) == 3:
+            elif len(offset) == _TZ_OFFSET_HH_LENGTH:
                 secs = int(offset[1:3]) * 3600
             if offset[0] == "-":
                 secs *= -1
@@ -681,9 +692,9 @@ def _parse_canonical_code(doc: Any, dummy0: Any) -> Code:
 def _parse_canonical_regex(doc: Any, dummy0: Any) -> Regex[str]:
     """Decode a JSON regex to bson.regex.Regex."""
     regex = doc["$regularExpression"]
-    if len(doc) != 1:
+    if len(doc) != _SINGLE_FIELD_COUNT:
         raise TypeError(f"Bad $regularExpression, extra field(s): {doc}")
-    if len(regex) != 2:
+    if len(regex) != _REGEX_COMPONENT_COUNT:
         raise TypeError(
             f'Bad $regularExpression must include only "pattern and "options" components: {doc}'
         )
@@ -718,7 +729,7 @@ def _parse_canonical_dbpointer(doc: Any, dummy0: Any) -> Any:
             raise TypeError(f"Bad $dbPointer, extra field $db: {dbref_doc}")
         if not isinstance(dbref.id, ObjectId):
             raise TypeError(f"Bad $dbPointer, $id must be an ObjectId: {dbref_doc}")
-        if len(dbref_doc) != 2:
+        if len(dbref_doc) != _DBREF_COMPONENT_COUNT:
             raise TypeError(f"Bad $dbPointer, extra field(s) in DBRef: {dbref_doc}")
         return dbref
     else:

bson/objectid.py

@@ -29,6 +29,8 @@
 from bson.tz_util import utc
 
 _MAX_COUNTER_VALUE = 0xFFFFFF
+_OBJECT_ID_BYTE_LENGTH = 12
+_OBJECT_ID_HEX_LENGTH = 24
 _PACK_INT = struct.Struct(">I").pack
 _PACK_INT_RANDOM = struct.Struct(">I5s").pack
 _UNPACK_INT = struct.Struct(">I").unpack
@@ -104,17 +106,17 @@ def __init__(self, oid: str | ObjectId | bytes | None = None) -> None:
 
             # 4 bytes current time, 5 bytes random, 3 bytes inc.
             self.__id = _PACK_INT_RANDOM(int(time.time()), ObjectId._random()) + _PACK_INT(inc)[1:4]
-        elif isinstance(oid, bytes) and len(oid) == 12:
+        elif isinstance(oid, bytes) and len(oid) == _OBJECT_ID_BYTE_LENGTH:
             self.__id = oid
         elif isinstance(oid, str):
-            if len(oid) == 24:
+            if len(oid) == _OBJECT_ID_HEX_LENGTH:
                 try:
                     self.__id = bytes.fromhex(oid)
                 except (TypeError, ValueError):
                     _raise_invalid_id(oid)
                 # bytes.fromhex() skips ASCII whitespace, so a 24-character
                 # string can decode to fewer than 12 bytes.
-                if len(self.__id) != 12:
+                if len(self.__id) != _OBJECT_ID_BYTE_LENGTH:
                     _raise_invalid_id(oid)
             else:
                 _raise_invalid_id(oid)

pymongo/_azure_helpers.py

@@ -20,6 +20,8 @@
 from typing import Any
 from urllib.parse import quote
 
+_HTTP_OK = 200
+
 
 def _get_azure_response(
     resource: str, client_id: str | None = None, timeout: float = 5
@@ -42,7 +44,7 @@ def _get_azure_response(
         msg = f"Failed to acquire IMDS access token: {e}"
         raise ValueError(msg) from None
 
-    if status != 200:
+    if status != _HTTP_OK:
         msg = "Failed to acquire IMDS access token."
         raise ValueError(msg)
     try: