Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Use _call within _get_file to ensure retries #384

Closed
wants to merge 7 commits into from
Closed

Use _call within _get_file to ensure retries #384

Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
360445e
Use _call with _get_file to ensure retries
Apr 29, 2021
dd21f4c
Fix formatting
Apr 30, 2021
72f1e1d
Delete code in _get_file that is duplicated in _get_args
Apr 30, 2021
f53a562
Pass kwargs from _get_file to _call
Apr 30, 2021
96cf889
get file in DEFAULT_BLOCK_SIZE chunks
Apr 30, 2021
7614876
Dont use break
Apr 30, 2021
ca9df18
define chunksize kwarg in _get_file
Apr 30, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 18 additions & 48 deletions gcsfs/core.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1141,18 +1141,10 @@ async def _isdir(self, path):
async def _find(self, path, withdirs=False, detail=False, prefix="", **kwargs):
path = self._strip_protocol(path)
bucket, key = self.split_path(path)
out, _ = await self._do_list_objects(
path,
delimiter=None,
prefix=prefix,
)
out, _ = await self._do_list_objects(path, delimiter=None, prefix=prefix)
if not out and key:
try:
out = [
await self._get_object(
path,
)
]
out = [await self._get_object(path)]
except FileNotFoundError:
out = []
dirs = []
Expand Down Expand Up @@ -1189,43 +1181,26 @@ async def _find(self, path, withdirs=False, detail=False, prefix="", **kwargs):
return {o["name"]: o for o in out}
return [o["name"] for o in out]

async def _get_file(self, rpath, lpath, **kwargs):
async def _get_file(self, rpath, lpath, chunksize=50 * 2 ** 20, **kwargs):
if await self._isdir(rpath):
return
u2 = self.url(rpath)
headers = kwargs.pop("headers", {})
consistency = kwargs.pop("consistency", self.consistency)
if "User-Agent" not in headers:
headers["User-Agent"] = "python-gcsfs/" + version
headers.update(self.heads or {}) # add creds

# needed for requester pays buckets
if self.requester_pays:
if isinstance(self.requester_pays, str):
user_project = self.requester_pays
else:
user_project = self.project
kwargs["userProject"] = user_project

async with self.session.get(
url=u2,
params=kwargs,
headers=headers,
timeout=self.requests_timeout,
) as r:
r.raise_for_status()
checker = get_consistency_checker(consistency)

os.makedirs(os.path.dirname(lpath), exist_ok=True)
with open(lpath, "wb") as f2:
while True:
data = await r.content.read(4096 * 32)
if not data:
break
f2.write(data)
checker.update(data)
checker = get_consistency_checker(consistency)
os.makedirs(os.path.dirname(lpath), exist_ok=True)

checker.validate_http_response(r)
bucket, key = self.split_path(rpath)
metadata = await self._call("GET", "b/{}/o/{}", bucket, key, json_out=True)
data_size = int(metadata["size"])
with open(lpath, "wb") as f2:
offset = 0
while offset < data_size:
head = {"Range": "bytes=%i-%i" % (offset, offset + chunksize - 1)}
headers, data = await self._call("GET", u2, headers=head, **kwargs)
f2.write(data)
checker.update(data)
offset += chunksize
checker.validate_headers(headers)

def _open(
self,
Expand Down Expand Up @@ -1313,12 +1288,7 @@ def validate_response(self, status, content, path, headers=None):
elif error:
raise HttpError(error)
elif status:
raise HttpError(
{
"code": status,
"message": msg,
}
) # text-like
raise HttpError({"code": status, "message": msg}) # text-like
else:
raise RuntimeError(msg)
else:
Expand Down