James Phoenix
08/14/2022, 9:21 PM@property
def basepath(self) -> str:
return f"gcs://{self.bucket_path}"
@property
def filesystem(self) -> RemoteFileSystem:
settings = {}
if self.service_account_info:
try:
settings["token"] = json.loads(
self.service_account_info.get_secret_value()
)
except json.JSONDecodeError:
raise ValueError(
"Unable to load provided service_account_info. Please make sure that the provided value is a valid JSON string."
)
remote_file_system = RemoteFileSystem(
basepath=f"gcs://{self.bucket_path}", settings=settings
)
return remote_file_system