Method info

pyathena/filesystem/s3.py:508–577

(self, path: str, **kwargs)

                break

    def info(self, path: str, **kwargs) -> S3Object:
        refresh = kwargs.pop("refresh", False)
        path = self._strip_protocol(path)
        bucket, key, path_version_id = self.parse_path(path)
        version_id = path_version_id if path_version_id else kwargs.pop("version_id", None)
        if path in ["/", ""]:
            return S3Object(
                init={
                    "ContentLength": 0,
                    "ContentType": None,
                    "StorageClass": S3StorageClass.S3_STORAGE_CLASS_BUCKET,
                    "ETag": None,
                    "LastModified": None,
                },
                type=S3ObjectType.S3_OBJECT_TYPE_DIRECTORY,
                bucket=bucket,
                key=None,
                version_id=None,
            )
        if not refresh:
            caches: list[S3Object] | S3Object | None = self._ls_from_cache(path)
            if caches is not None:
                if isinstance(caches, list):
                    cache = next((c for c in caches if c.name == path), None)
                elif caches.name == path:
                    cache = caches
                else:
                    cache = None

                if cache:
                    if (
                        self.version_aware
                        and not version_id
                        and cache.get("type") == S3ObjectType.S3_OBJECT_TYPE_FILE
                        and not cache.get("version_id")
                    ):
                        # A version-aware lookup needs the version to pin;
                        # treat a version-less cached entry (e.g., populated
                        # by a listing) as stale and head the object again.
                        refresh = True
                    else:
                        return cache
                else:
                    return self._directory_object(
                        bucket, key.rstrip("/") if key else None, version_id
                    )
        if key:
            object_info = self._head_object(path, refresh=refresh, version_id=version_id)
            if object_info:
                return object_info
        else:
            bucket_info = self._head_bucket(path, refresh=refresh)
            if bucket_info:
                return bucket_info
            raise FileNotFoundError(path)

        response = self._call(
            self._client.list_objects_v2,