Check the dircache for a cached entry of the path. fsspec's implementation assumes every dircache value is a listing, but S3FileSystem also caches a single S3Object under the object's own path (HeadObject/HeadBucket results). Guard the parent lookup so that looking u
(self, path: str)
| 1740 | path = self._parent(path) |
| 1741 | |
| 1742 | def _ls_from_cache(self, path: str) -> list[S3Object] | S3Object | None: |
| 1743 | """Check the dircache for a cached entry of the path. |
| 1744 | |
| 1745 | fsspec's implementation assumes every dircache value is a listing, |
| 1746 | but S3FileSystem also caches a single S3Object under the object's own |
| 1747 | path (HeadObject/HeadBucket results). Guard the parent lookup so that |
| 1748 | looking up a child path of a cached object does not fail, and fall |
| 1749 | through to the S3 API instead. |
| 1750 | """ |
| 1751 | cache = self.dircache.get(path.rstrip("/")) |
| 1752 | if cache is not None: |
| 1753 | return cast("list[S3Object] | S3Object", cache) |
| 1754 | parent_cache = self.dircache.get(self._parent(path)) |
| 1755 | if isinstance(parent_cache, list): |
| 1756 | files = [ |
| 1757 | f |
| 1758 | for f in parent_cache |
| 1759 | if f["name"] == path |
| 1760 | or ( |
| 1761 | f["name"] == path.rstrip("/") |
| 1762 | and f["type"] == S3ObjectType.S3_OBJECT_TYPE_DIRECTORY |
| 1763 | ) |
| 1764 | ] |
| 1765 | if files: |
| 1766 | return files |
| 1767 | raise FileNotFoundError(path) |
| 1768 | return None |
| 1769 | |
| 1770 | def _open( |
| 1771 | self, |