Source code for neosqlite.binary

from __future__ import annotations

import base64


[docs] class Binary(bytes): """ A BSON Binary-like class for representing binary data in neosqlite. This class provides a PyMongo-compatible interface for storing binary data directly in documents (outside of GridFS). The data is automatically encoded/decoded when stored in SQLite JSON documents. Example usage: # Create binary data binary_data = Binary(b"some binary content") # Store in document collection.insert_one({"data": binary_data}) # Retrieve and use doc = collection.find_one({}) retrieved_data = doc["data"] # Returns Binary instance raw_bytes = bytes(retrieved_data) # Convert to bytes if needed """ # Type declaration for mypy _subtype: int # Binary subtypes BINARY_SUBTYPE = 0 FUNCTION_SUBTYPE = 1 OLD_BINARY_SUBTYPE = 2 UUID_SUBTYPE = 4 MD5_SUBTYPE = 5 COLUMN_SUBTYPE = 7 SENSITIVE_SUBTYPE = 8 VECTOR_SUBTYPE = 9 USER_DEFINED_SUBTYPE = 128 def __new__( cls, data: bytes | bytearray | memoryview, subtype: int = BINARY_SUBTYPE, ): """ Create a new Binary instance. Args: data: Binary data (bytes, bytearray, or memoryview) subtype: Binary subtype (default: BINARY_SUBTYPE) Returns: A new Binary instance """ if isinstance(data, (bytearray, memoryview)): data = bytes(data) elif not isinstance(data, bytes): raise TypeError("data must be bytes, bytearray, or memoryview") if not isinstance(subtype, int): raise TypeError("subtype must be an integer") if not (0 <= subtype < 256): raise ValueError("subtype must be in range [0, 256)") instance = super().__new__(cls, data) instance._subtype = subtype return instance @property def subtype(self) -> int: """Get the binary subtype.""" return self._subtype
[docs] def encode_for_storage(self) -> dict: """ Encode the binary data for JSON storage. Returns: A dictionary representation for JSON storage """ return { "__neosqlite_binary__": True, "data": base64.b64encode(self).decode("utf-8"), "subtype": self._subtype, }
[docs] @classmethod def decode_from_storage(cls, encoded_data: dict) -> Binary: """ Decode binary data from JSON storage. Args: encoded_data: Dictionary representation from JSON storage Returns: A Binary instance """ if ( not isinstance(encoded_data, dict) or "__neosqlite_binary__" not in encoded_data ): raise ValueError("Invalid encoded binary data") if "data" not in encoded_data: raise ValueError( "Invalid encoded binary data: missing 'data' field" ) data = base64.b64decode(encoded_data["data"]) subtype = encoded_data.get("subtype", cls.BINARY_SUBTYPE) return cls(data, subtype)
[docs] @classmethod def from_uuid(cls, uuid_value, uuid_representation=None): """ Create a Binary instance from a UUID. Args: uuid_value: A UUID instance uuid_representation: UUID representation (ignored in neosqlite) Returns: A Binary instance with UUID_SUBTYPE """ import uuid if not isinstance(uuid_value, uuid.UUID): raise TypeError("uuid_value must be a UUID instance") # Convert UUID to bytes return cls(uuid_value.bytes, cls.UUID_SUBTYPE)
[docs] def as_uuid(self, uuid_representation=None): """ Convert this Binary instance to a UUID. Returns: A UUID instance Raises: ValueError: If the subtype is not UUID_SUBTYPE """ if self._subtype != self.UUID_SUBTYPE: raise ValueError("Binary subtype is not UUID_SUBTYPE") import uuid return uuid.UUID(bytes=bytes(self))
def __repr__(self) -> str: """Return a string representation of the Binary instance.""" if self._subtype == self.BINARY_SUBTYPE: return f"Binary({super().__repr__()})" else: return f"Binary({super().__repr__()}, {self._subtype})" def __eq__(self, other) -> bool: """Check equality with another object.""" if isinstance(other, Binary): return super().__eq__(other) and self._subtype == other._subtype elif isinstance(other, bytes): return ( super().__eq__(other) and self._subtype == self.BINARY_SUBTYPE ) return False def __ne__(self, other) -> bool: """Check inequality with another object.""" return not self.__eq__(other) def __hash__(self) -> int: """Return a hash value for the Binary instance.""" return hash((bytes(self), self._subtype))