Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
domain
URI of the domain name to access. E.g.: tall.data.hdfgroup.org.
mode
Access mode: 'r', 'r+', 'w', or 'a'
endpoint
Server endpoint. Defaults to "http://localhost:5000"
linked_domain
Create new domain using the root of the linked domain
"""
groupid = None
# if we're passed a GroupId as domain, just initialize the file object
# with that. This will be faster and enable the File object to share the same http connection.
if mode is None and endpoint is None and username is None \
and password is None and isinstance(domain, GroupID):
groupid = domain
else:
if mode and mode not in ('r', 'r+', 'w', 'w-', 'x', 'a'):
raise ValueError(
"Invalid mode; must be one of r, r+, w, w-, x, a")
if mode is None:
mode = 'r'
cfg = Config() # pulls in state from a .hscfg file (if found).
# accept domain values in the form:
# http://server:port/home/user/myfile.h5
# or
# https://server:port/home/user/myfile.h5
# or
def __init__(self, parent):
""" Private constructor.
"""
self._parent = parent
if isinstance(parent.id, GroupID):
self._req_prefix = "/groups/" + parent.id.uuid + "/attributes/"
elif isinstance(parent.id, TypeID):
self._req_prefix = "/datatypes/" + parent.id.uuid + "/attributes/"
elif isinstance(parent.id, DatasetID):
self._req_prefix = "/datasets/" + parent.id.uuid + "/attributes/"
else:
# "unknown id"
self._req_prefix = ""
objdb = self._parent.id.http_conn.getObjDb()
if objdb:
# _objdb is meta-data pulled from the domain on open.
# see if we can extract the link json from there
objid = self._parent.id.uuid
if objid not in objdb:
raise IOError("Expected to find {} in objdb".format(objid))
obj_json = objdb[objid]
req = "/groups/" + parent_uuid + "/links/" + link
try:
rsp_json = self.GET(req)
except IOError as ioe:
self.log.debug("Got ioe: {}".format(ioe))
create_group = True
if create_group:
link_json = {'id': parent_uuid, 'name': link}
body = {'link': link_json }
self.log.debug("create group with body: {}".format(body))
rsp = self.POST('/groups', body=body)
group_json = rsp
groupId = GroupID(self, group_json)
sub_group = Group(groupId)
if parent_name:
if parent_name[-1] == '/':
parent_name = parent_name + link
else:
parent_name = parent_name + '/' + link
self.log.debug("create group - parent name: {}".format(parent_name))
sub_group._name = parent_name
parent_uuid = sub_group.id.id
else:
# sub-group already exsits
self.log.debug("create group - found subgroup: {}".format(link))
if "link" not in rsp_json:
raise IOError("Unexpected Error")
link_json = rsp_json["link"]
if link_json["class"] != 'H5L_TYPE_HARD':
http_conn._objdb = objdb
if root_uuid in objdb:
group_json = objdb[root_uuid]
if not group_json:
# get the group json for the root group
req = "/groups/" + root_uuid
rsp = http_conn.GET(req)
if rsp.status_code != 200:
http_conn.close()
raise IOError(rsp.status_code, "Unexpected Error")
group_json = json.loads(rsp.text)
groupid = GroupID(None, group_json, http_conn=http_conn)
# end else
self._name = '/'
self._id = groupid
self._verboseInfo = None # aditional state we'll get when requested
self._verboseUpdated = None # when the verbose data was fetched
Group.__init__(self, self._id)
def file(self):
""" Return a File instance associated with this object """
from .files import File
http_conn = self._id.http_conn
root_uuid = http_conn.root_uuid
# construct a group json, so we don't need to do a request
group_json = {}
group_json["root"] = root_uuid
group_json["id"] = root_uuid
group_json["domain"] = http_conn.domain
group_json["created"] = http_conn.created
group_json["lastModified"] = http_conn.modified
groupid = GroupID(None, group_json, http_conn=http_conn)
return File(groupid)