Iterates over a file's chunks using a single cursor. Raises CorruptGridFile when encountering any truncated, missing, or extra chunk in a file.
| 1808 | |
| 1809 | |
| 1810 | class GridOutChunkIterator: |
| 1811 | """Iterates over a file's chunks using a single cursor. |
| 1812 | |
| 1813 | Raises CorruptGridFile when encountering any truncated, missing, or extra |
| 1814 | chunk in a file. |
| 1815 | """ |
| 1816 | |
| 1817 | def __init__( |
| 1818 | self, |
| 1819 | grid_out: GridOut, |
| 1820 | chunks: Collection[Any], |
| 1821 | session: Optional[ClientSession], |
| 1822 | next_chunk: Any, |
| 1823 | ) -> None: |
| 1824 | self._id = grid_out._id |
| 1825 | self._chunk_size = int(grid_out.chunk_size) |
| 1826 | self._length = int(grid_out.length) |
| 1827 | self._chunks = chunks |
| 1828 | self._session = session |
| 1829 | self._next_chunk = next_chunk |
| 1830 | self._num_chunks = math.ceil(float(self._length) / self._chunk_size) |
| 1831 | self._cursor = None |
| 1832 | |
| 1833 | _cursor: Optional[Cursor[Any]] |
| 1834 | |
| 1835 | def expected_chunk_length(self, chunk_n: int) -> int: |
| 1836 | if chunk_n < self._num_chunks - 1: |
| 1837 | return self._chunk_size |
| 1838 | return self._length - (self._chunk_size * (self._num_chunks - 1)) |
| 1839 | |
| 1840 | def __iter__(self) -> GridOutChunkIterator: |
| 1841 | return self |
| 1842 | |
| 1843 | def _create_cursor(self) -> None: |
| 1844 | filter = {"files_id": self._id} |
| 1845 | if self._next_chunk > 0: |
| 1846 | filter["n"] = {"$gte": self._next_chunk} |
| 1847 | _disallow_transactions(self._session) |
| 1848 | self._cursor = self._chunks.find(filter, sort=[("n", 1)], session=self._session) |
| 1849 | |
| 1850 | def _next_with_retry(self) -> Mapping[str, Any]: |
| 1851 | """Return the next chunk and retry once on CursorNotFound. |
| 1852 | |
| 1853 | We retry on CursorNotFound to maintain backwards compatibility in |
| 1854 | cases where two calls to read occur more than 10 minutes apart (the |
| 1855 | server's default cursor timeout). |
| 1856 | """ |
| 1857 | if self._cursor is None: |
| 1858 | self._create_cursor() |
| 1859 | assert self._cursor is not None |
| 1860 | try: |
| 1861 | return self._cursor.next() |
| 1862 | except CursorNotFound: |
| 1863 | self._cursor.close() |
| 1864 | self._create_cursor() |
| 1865 | return self._cursor.next() |
| 1866 | |
| 1867 | def next(self) -> Mapping[str, Any]: |