def put_bytes()

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

• 23 lines of code
• 5 McCabe index (conditional complexity)

    def put_bytes(self, content: bytes) -> Optional[SharedMemoryMetadata]:
        """
        Writes the given bytes into shared memory.
        Returns metadata about the shared memory region to which the content was
        written if successful, None otherwise.
        """
        if content is None:
            return None
        mem_map_name = str(uuid.uuid4())
        content_length = len(content)
        shared_mem_map = self._create(mem_map_name, content_length)
        if shared_mem_map is None:
            return None
        try:
            num_bytes_written = shared_mem_map.put_bytes(content)
        except Exception as e:
            logger.warning('Cannot write %s bytes into shared memory %s - %s',
                           content_length, mem_map_name, e)
            shared_mem_map.dispose()
            return None
        if num_bytes_written != content_length:
            logger.error(
                'Cannot write data into shared memory %s (%s != %s)',
                mem_map_name, num_bytes_written, content_length)
            shared_mem_map.dispose()
            return None
        self.allocated_mem_maps[mem_map_name] = shared_mem_map
        return SharedMemoryMetadata(mem_map_name, content_length)