(self)
| 69 | return False |
| 70 | |
| 71 | def _close(self) -> None: |
| 72 | result: Buffer |
| 73 | if self.__callback: |
| 74 | if self.__buf.tell() == 0: |
| 75 | # Empty file: |
| 76 | result = b"" |
| 77 | else: |
| 78 | # Return the data without actually loading it into memory, |
| 79 | # relying on Python's buffer API and mmap(). mmap() just gives |
| 80 | # a view directly into the filesystem's memory cache, so it |
| 81 | # doesn't result in duplicate memory use. |
| 82 | self.__buf.seek(0, 0) |
| 83 | result = memoryview( |
| 84 | mmap.mmap(self.__buf.fileno(), 0, access=mmap.ACCESS_READ) |
| 85 | ) |
| 86 | self.__callback(result) |
| 87 | |
| 88 | # We assign this to None here, because otherwise we can get into |
| 89 | # really tricky problems where the CPython interpreter dead locks |
| 90 | # because the callback is holding a reference to something which |
| 91 | # has a __del__ method. Setting this to None breaks the cycle |
| 92 | # and allows the garbage collector to do it's thing normally. |
| 93 | self.__callback = None |
| 94 | |
| 95 | # Closing the temporary file releases memory and frees disk space. |
| 96 | # Important when caching big files. |
| 97 | self.__buf.close() |
| 98 | |
| 99 | def read(self, amt: int | None = None) -> bytes: |
| 100 | data: bytes = self.__fp.read(amt) |
no test coverage detected