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