def put_bytes()

in azure_functions_worker/bindings/shared_memory_data_transfer/shared_memory_map.py [0:0]

• 19 lines of code
• 3 McCabe index (conditional complexity)

    def put_bytes(self, content: bytes) -> Optional[int]:
        """
        Writes the given content bytes into this SharedMemoryMap.
        The number of bytes written must be less than or equal to the size of
        the SharedMemoryMap.
        Returns the number of bytes of content written.
        """
        if content is None:
            return None
        content_length = len(content)
        # Seek past the MemoryMapInitialized flag section of the header
        self.mem_map.seek(consts.MEM_MAP_INITIALIZED_FLAG_NUM_BYTES)
        # Write the content length into the header
        content_length_bytes = content_length.to_bytes(
            consts.CONTENT_LENGTH_NUM_BYTES, byteorder=sys.byteorder)
        num_content_length_bytes = len(content_length_bytes)
        num_content_length_bytes_written = self.mem_map.write(
            content_length_bytes)
        if num_content_length_bytes_written != num_content_length_bytes:
            logger.error(
                'Cannot write content size to memory map %s (%s != %s)',
                self.mem_map_name, num_content_length_bytes_written,
                num_content_length_bytes)
            return 0
        # Write the content
        num_content_bytes_written = self.mem_map.write(content)
        self.mem_map.flush()
        return num_content_bytes_written