Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
help="Path to the GCP token", allow_none=True, default_value=None
).tag(config=True, env="JPYNB_GCS_TOKEN_PATH")
region_name = Unicode("us-east-1", help="Region name").tag(
config=True, env="JPYNB_GCS_REGION_NAME"
)
bucket = Unicode("notebooks", help="Bucket name to store notebooks").tag(
config=True, env="JPYNB_GCS_BUCKET"
)
prefix = Unicode("", help="Prefix path inside the specified bucket").tag(
config=True
)
separator = Unicode("/", help="Path separator").tag(config=True)
dir_keep_file = Unicode(
".gcskeep", help="Empty file to create when creating directories"
).tag(config=True)
def __init__(self, log, **kwargs):
super(GCSFS, self).__init__(**kwargs)
self.log = log
token = os.path.expanduser(self.token)
self.fs = gcsfs.GCSFileSystem(project=self.project, token=token)
self.init()
def init(self):
self.mkdir("")
self.ls("")
assert self.isdir(""), "The root directory should exists :)"
from s3contents.gcs_fs import GCSFS
from s3contents.genericmanager import GenericContentsManager
from s3contents.ipycompat import Unicode
class GCSContentsManager(GenericContentsManager):
project = Unicode(help="GCP Project", allow_none=True, default_value=None).tag(
config=True, env="JPYNB_GCS_PROJECT"
)
token = Unicode(
help="Path to the GCP token", allow_none=True, default_value=None
).tag(config=True, env="JPYNB_GCS_TOKEN_PATH")
region_name = Unicode("us-east-1", help="Region name").tag(
config=True, env="JPYNB_GCS_REGION_NAME"
)
bucket = Unicode("notebooks", help="Bucket name to store notebooks").tag(
config=True, env="JPYNB_GCS_BUCKET"
)
prefix = Unicode("", help="Prefix path inside the specified bucket").tag(
config=True
)
separator = Unicode("/", help="Path separator").tag(config=True)
def __init__(self, *args, **kwargs):
super(GCSContentsManager, self).__init__(*args, **kwargs)
self._fs = GCSFS(
log=self.log,
secret_access_key = Unicode(
help="S3/AWS secret access key", allow_none=True, default_value=None
).tag(config=True, env="JPYNB_S3_SECRET_ACCESS_KEY")
endpoint_url = Unicode("s3.amazonaws.com", help="S3 endpoint URL").tag(
config=True, env="JPYNB_S3_ENDPOINT_URL"
)
region_name = Unicode("us-east-1", help="Region name").tag(
config=True, env="JPYNB_S3_REGION_NAME"
)
bucket = Unicode("notebooks", help="Bucket name to store notebooks").tag(
config=True, env="JPYNB_S3_BUCKET"
)
signature_version = Unicode(help="").tag(config=True)
sse = Unicode(help="Type of server-side encryption to use").tag(config=True)
kms_key_id = Unicode(help="KMS ID to use to encrypt workbooks").tag(config=True)
prefix = Unicode("", help="Prefix path inside the specified bucket").tag(
config=True
)
delimiter = Unicode("/", help="Path delimiter").tag(config=True)
dir_keep_file = Unicode(
".s3keep", help="Empty file to create when creating directories"
).tag(config=True)
session_token = Unicode(
help="S3/AWS session token", allow_none=True, default_value=None
).tag(config=True, env="JPYNB_S3_SESSION_TOKEN")
boto3_session = Any(
help="Place to store customer boto3 session instance - likely passed in"
GenericFileCheckpoints,
HasTraits,
Unicode,
from_dict,
reads,
)
DUMMY_CREATED_DATE = datetime.datetime.fromtimestamp(86400)
NBFORMAT_VERSION = 4
class GenericContentsManager(ContentsManager, HasTraits):
# This makes the checkpoints get saved on this directory
root_dir = Unicode("./", config=True)
def __init__(self, *args, **kwargs):
super(GenericContentsManager, self).__init__(*args, **kwargs)
self._fs = None
def get_fs(self):
return self._fs
fs = property(get_fs)
def _checkpoints_class_default(self):
return GenericFileCheckpoints
def do_error(self, msg, code=500):
raise HTTPError(code, msg)
config=True, env="JPYNB_S3_ENDPOINT_URL"
)
region_name = Unicode("us-east-1", help="Region name").tag(
config=True, env="JPYNB_S3_REGION_NAME"
)
bucket = Unicode("notebooks", help="Bucket name to store notebooks").tag(
config=True, env="JPYNB_S3_BUCKET"
)
signature_version = Unicode(help="").tag(config=True)
sse = Unicode(help="Type of server-side encryption to use").tag(config=True)
kms_key_id = Unicode(help="KMS ID to use to encrypt workbooks").tag(config=True)
prefix = Unicode("", help="Prefix path inside the specified bucket").tag(
config=True
)
delimiter = Unicode("/", help="Path delimiter").tag(config=True)
dir_keep_file = Unicode(
".s3keep", help="Empty file to create when creating directories"
).tag(config=True)
session_token = Unicode(
help="S3/AWS session token", allow_none=True, default_value=None
).tag(config=True, env="JPYNB_S3_SESSION_TOKEN")
boto3_session = Any(
help="Place to store customer boto3 session instance - likely passed in"
)
def __init__(self, log, **kwargs):
super(S3FS, self).__init__(**kwargs)
self.log = log
class S3FS(GenericFS):
access_key_id = Unicode(
help="S3/AWS access key ID", allow_none=True, default_value=None
).tag(config=True, env="JPYNB_S3_ACCESS_KEY_ID")
secret_access_key = Unicode(
help="S3/AWS secret access key", allow_none=True, default_value=None
).tag(config=True, env="JPYNB_S3_SECRET_ACCESS_KEY")
endpoint_url = Unicode("s3.amazonaws.com", help="S3 endpoint URL").tag(
config=True, env="JPYNB_S3_ENDPOINT_URL"
)
region_name = Unicode("us-east-1", help="Region name").tag(
config=True, env="JPYNB_S3_REGION_NAME"
)
bucket = Unicode("notebooks", help="Bucket name to store notebooks").tag(
config=True, env="JPYNB_S3_BUCKET"
)
signature_version = Unicode(help="").tag(config=True)
sse = Unicode(help="Type of server-side encryption to use").tag(config=True)
kms_key_id = Unicode(help="KMS ID to use to encrypt workbooks").tag(config=True)
prefix = Unicode("", help="Prefix path inside the specified bucket").tag(
config=True
)
delimiter = Unicode("/", help="Path delimiter").tag(config=True)
dir_keep_file = Unicode(
".s3keep", help="Empty file to create when creating directories"
).tag(config=True)
session_token = Unicode(
"""
class S3FS(GenericFS):
access_key_id = Unicode(
help="S3/AWS access key ID", allow_none=True, default_value=None
).tag(config=True, env="JPYNB_S3_ACCESS_KEY_ID")
secret_access_key = Unicode(
help="S3/AWS secret access key", allow_none=True, default_value=None
).tag(config=True, env="JPYNB_S3_SECRET_ACCESS_KEY")
endpoint_url = Unicode("s3.amazonaws.com", help="S3 endpoint URL").tag(
config=True, env="JPYNB_S3_ENDPOINT_URL"
)
region_name = Unicode("us-east-1", help="Region name").tag(
config=True, env="JPYNB_S3_REGION_NAME"
)
bucket = Unicode("notebooks", help="Bucket name to store notebooks").tag(
config=True, env="JPYNB_S3_BUCKET"
)
signature_version = Unicode(help="").tag(config=True)
sse = Unicode(help="Type of server-side encryption to use").tag(config=True)
kms_key_id = Unicode(help="KMS ID to use to encrypt workbooks").tag(config=True)
prefix = Unicode("", help="Prefix path inside the specified bucket").tag(
config=True
)
delimiter = Unicode("/", help="Path delimiter").tag(config=True)
dir_keep_file = Unicode(
".s3keep", help="Empty file to create when creating directories"
help="S3/AWS access key ID", allow_none=True, default_value=None).tag(
config=True, env="JPYNB_S3_ACCESS_KEY_ID")
secret_access_key = Unicode(
help="S3/AWS secret access key", allow_none=True, default_value=None).tag(
config=True, env="JPYNB_S3_SECRET_ACCESS_KEY")
endpoint_url = Unicode(
"s3.amazonaws.com", help="S3 endpoint URL").tag(
config=True, env="JPYNB_S3_ENDPOINT_URL")
region_name = Unicode(
"us-east-1", help="Region Name").tag(
config=True, env="JPYNB_S3_REGION_NAME")
bucket_name = Unicode(
"notebooks", help="Bucket name to store notebooks").tag(
config=True, env="JPYNB_S3_BUCKET_NAME")
prefix = Unicode("", help="Prefix path inside the specified bucket").tag(config=True)
signature_version = Unicode(help="").tag(config=True)
delimiter = Unicode("/", help="Path delimiter").tag(config=True)
dir_keep_file = Unicode(
".s3keep", help="Empty file to create when creating directories").tag(config=True)
def __init__(self, log, **kwargs):
super(S3FS, self).__init__(**kwargs)
self.log = log
config = None
if self.signature_version:
config = Config(signature_version=self.signature_version)
self.client = boto3.client(
"s3",
region_name = Unicode("us-east-1", help="Region name").tag(
config=True, env="JPYNB_S3_REGION_NAME"
)
bucket = Unicode("notebooks", help="Bucket name to store notebooks").tag(
config=True, env="JPYNB_S3_BUCKET"
)
signature_version = Unicode(help="").tag(config=True)
sse = Unicode(help="Type of server-side encryption to use").tag(config=True)
kms_key_id = Unicode(help="KMS ID to use to encrypt workbooks").tag(config=True)
prefix = Unicode("", help="Prefix path inside the specified bucket").tag(
config=True
)
delimiter = Unicode("/", help="Path delimiter").tag(config=True)
dir_keep_file = Unicode(
".s3keep", help="Empty file to create when creating directories"
).tag(config=True)
session_token = Unicode(
help="S3/AWS session token", allow_none=True, default_value=None
).tag(config=True, env="JPYNB_S3_SESSION_TOKEN")
boto3_session = Any(
help="Place to store customer boto3 session instance - likely passed in"
)
def __init__(self, log, **kwargs):
super(S3FS, self).__init__(**kwargs)
self.log = log
client_kwargs = {
"""
Utilities to make S3 look like a regular file system
"""
import six
import boto3
from botocore.client import Config
from s3contents.ipycompat import HasTraits, Unicode
class S3FS(HasTraits):
access_key_id = Unicode(
help="S3/AWS access key ID", allow_none=True, default_value=None).tag(
config=True, env="JPYNB_S3_ACCESS_KEY_ID")
secret_access_key = Unicode(
help="S3/AWS secret access key", allow_none=True, default_value=None).tag(
config=True, env="JPYNB_S3_SECRET_ACCESS_KEY")
endpoint_url = Unicode(
"s3.amazonaws.com", help="S3 endpoint URL").tag(
config=True, env="JPYNB_S3_ENDPOINT_URL")
region_name = Unicode(
"us-east-1", help="Region Name").tag(
config=True, env="JPYNB_S3_REGION_NAME")
bucket_name = Unicode(
"notebooks", help="Bucket name to store notebooks").tag(
config=True, env="JPYNB_S3_BUCKET_NAME")
prefix = Unicode("", help="Prefix path inside the specified bucket").tag(config=True)
signature_version = Unicode(help="").tag(config=True)
delimiter = Unicode("/", help="Path delimiter").tag(config=True)