1""" 2BitBake 'Fetch' implementation for Google Cloup Platform Storage. 3 4Class for fetching files from Google Cloud Storage using the 5Google Cloud Storage Python Client. The GCS Python Client must 6be correctly installed, configured and authenticated prior to use. 7Additionally, gsutil must also be installed. 8 9""" 10 11# Copyright (C) 2023, Snap Inc. 12# 13# Based in part on bb.fetch2.s3: 14# Copyright (C) 2017 Andre McCurdy 15# 16# SPDX-License-Identifier: GPL-2.0-only 17# 18# Based on functions from the base bb module, Copyright 2003 Holger Schurig 19 20import os 21import bb 22import urllib.parse, urllib.error 23from bb.fetch2 import FetchMethod 24from bb.fetch2 import FetchError 25from bb.fetch2 import logger 26 27class GCP(FetchMethod): 28 """ 29 Class to fetch urls via GCP's Python API. 30 """ 31 def __init__(self): 32 self.gcp_client = None 33 34 def supports(self, ud, d): 35 """ 36 Check to see if a given url can be fetched with GCP. 37 """ 38 return ud.type in ['gs'] 39 40 def recommends_checksum(self, urldata): 41 return True 42 43 def urldata_init(self, ud, d): 44 if 'downloadfilename' in ud.parm: 45 ud.basename = ud.parm['downloadfilename'] 46 else: 47 ud.basename = os.path.basename(ud.path) 48 49 ud.localfile = d.expand(urllib.parse.unquote(ud.basename)) 50 51 def get_gcp_client(self): 52 from google.cloud import storage 53 self.gcp_client = storage.Client(project=None) 54 55 def download(self, ud, d): 56 """ 57 Fetch urls using the GCP API. 58 Assumes localpath was called first. 59 """ 60 logger.debug2(f"Trying to download gs://{ud.host}{ud.path} to {ud.localpath}") 61 if self.gcp_client is None: 62 self.get_gcp_client() 63 64 bb.fetch2.check_network_access(d, "gsutil stat", ud.url) 65 66 # Path sometimes has leading slash, so strip it 67 path = ud.path.lstrip("/") 68 blob = self.gcp_client.bucket(ud.host).blob(path) 69 blob.download_to_filename(ud.localpath) 70 71 # Additional sanity checks copied from the wget class (although there 72 # are no known issues which mean these are required, treat the GCP API 73 # tool with a little healthy suspicion). 74 if not os.path.exists(ud.localpath): 75 raise FetchError(f"The GCP API returned success for gs://{ud.host}{ud.path} but {ud.localpath} doesn't exist?!") 76 77 if os.path.getsize(ud.localpath) == 0: 78 os.remove(ud.localpath) 79 raise FetchError(f"The downloaded file for gs://{ud.host}{ud.path} resulted in a zero size file?! Deleting and failing since this isn't right.") 80 81 return True 82 83 def checkstatus(self, fetch, ud, d): 84 """ 85 Check the status of a URL. 86 """ 87 logger.debug2(f"Checking status of gs://{ud.host}{ud.path}") 88 if self.gcp_client is None: 89 self.get_gcp_client() 90 91 bb.fetch2.check_network_access(d, "gsutil stat", ud.url) 92 93 # Path sometimes has leading slash, so strip it 94 path = ud.path.lstrip("/") 95 if self.gcp_client.bucket(ud.host).blob(path).exists() == False: 96 raise FetchError(f"The GCP API reported that gs://{ud.host}{ud.path} does not exist") 97 else: 98 return True 99