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 :)"
def git_versions_from_keywords(keywords, tag_prefix, verbose):
"""Get version information from git keywords."""
if not keywords:
raise NotThisMethod("no keywords at all, weird")
refnames = keywords["refnames"].strip()
if refnames.startswith("$Format"):
if verbose:
print("keywords are unexpanded, not using")
raise NotThisMethod("unexpanded keywords, not a git-archive tarball")
refs = set([r.strip() for r in refnames.strip("()").split(",")])
# starting in git-1.8.3, tags are listed as "tag: foo-1.0" instead of
# just "foo-1.0". If we see a "tag: " prefix, prefer those.
TAG = "tag: "
tags = set([r[len(TAG):] for r in refs if r.startswith(TAG)])
if not tags:
# Either we're using git < 1.8.3, or there really are no tags. We use
# a heuristic: assume all version tags have a digit. The old git %d
# expansion behaves like git log --decorate=short and strips out the
# refs/heads/ and refs/tags/ prefixes that would let us distinguish
# between branches and tags. By ignoring refnames without digits, we
def git_versions_from_keywords(keywords, tag_prefix, verbose):
"""Get version information from git keywords."""
if not keywords:
raise NotThisMethod("no keywords at all, weird")
refnames = keywords["refnames"].strip()
if refnames.startswith("$Format"):
if verbose:
print("keywords are unexpanded, not using")
raise NotThisMethod("unexpanded keywords, not a git-archive tarball")
refs = set([r.strip() for r in refnames.strip("()").split(",")])
# starting in git-1.8.3, tags are listed as "tag: foo-1.0" instead of
# just "foo-1.0". If we see a "tag: " prefix, prefer those.
TAG = "tag: "
tags = set([r[len(TAG):] for r in refs if r.startswith(TAG)])
if not tags:
# Either we're using git < 1.8.3, or there really are no tags. We use
# a heuristic: assume all version tags have a digit. The old git %d
# expansion behaves like git log --decorate=short and strips out the
# refs/heads/ and refs/tags/ prefixes that would let us distinguish
# between branches and tags. By ignoring refnames without digits, we
# filter out many common branch names like "release" and
# "stabilization", as well as "HEAD" and "master".
tags = set([r for r in refs if re.search(r'\d', r)])
if verbose:
print("discarding '%s', no digits" % ",".join(refs - tags))
def get_versions():
"""Get version information or return default if unable to do so."""
# I am in _version.py, which lives at ROOT/VERSIONFILE_SOURCE. If we have
# __file__, we can work backwards from there to the root. Some
# py2exe/bbfreeze/non-CPython implementations don't do __file__, in which
# case we can only use expanded keywords.
cfg = get_config()
verbose = cfg.verbose
try:
return git_versions_from_keywords(get_keywords(), cfg.tag_prefix, verbose)
except NotThisMethod:
pass
try:
root = os.path.realpath(__file__)
# versionfile_source is the relative path from the top of the source
# tree (where the .git directory might live) to this file. Invert
# this to find the root from __file__.
for i in cfg.versionfile_source.split('/'):
root = os.path.dirname(root)
except NameError:
return {
"version": "0+unknown",
"full-revisionid": None,
"dirty": None,
"error": "unable to find root of source tree"
}
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()
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(