diff --git a/src/solid/solid_api.py b/src/solid/solid_api.py index 9eda911..52fb28c 100644 --- a/src/solid/solid_api.py +++ b/src/solid/solid_api.py @@ -5,7 +5,7 @@ from httpx import Response, HTTPStatusError from solid.auth import Auth -from solid.utils.api_util import get_root_url, LINK, get_parent_url, get_item_name +from solid.utils.api_util import get_root_url, LINK, get_parent_url, get_item_name, get_links_from_response from solid.utils.folder_utils import parse_folder_response @@ -215,8 +215,26 @@ def read_folder(self, url, options: ReadFolderOptions = ReadFolderOptions()) -> return parsed_folder - def get_item_links(self, url, options: Dict = None) -> Response: - raise Exception('Not implemented') + def get_item_links(self, url, options: Dict = {}) -> Links: + if not self.item_exists(url): + raise Exception(f'Item not found: {url}') + + options.update({ + "links": LINKS.INCLUDE_POSSIBLE, + "withAcl": True, + "withMeta": True, + }) + + if (options['links'] == LINKS.EXCLUDE): + raise Exception(f'Invalid option LINKS.EXCLUDE for getItemLinks') + + response = self.head(url) + links = get_links_from_response(response) + if (options['links'] == LINKS.INCLUDE): + if (not options.withAcl): del links['acl'] + if (not options.withMeta): del links['meta'] + + return links def copy_file(self, _from, to, options: WriteOptions = None) -> Response: raise Exception('Not implemented') diff --git a/src/solid/utils/api_util.py b/src/solid/utils/api_util.py index caefd5b..a36567c 100644 --- a/src/solid/utils/api_util.py +++ b/src/solid/utils/api_util.py @@ -1,5 +1,6 @@ from enum import Enum -from typing import List +from typing import List, Dict +import re class LINK(Enum): @@ -59,3 +60,107 @@ def are_folders(urls: List) -> bool: def are_files(urls: List) -> bool: pass + +def get_links_from_response(response) -> dict: + link_header = response.headers.get('link') + if (not link_header) or (link_header == ''): + return {} + else: + return parse_link_header(link_header, response.url) + +def parse_link_header(link_header: str, item_url) -> Dict: + link_dict = {} + link_header_list = parse_link_header_to_array(link_header) + if len(link_header_list) > 0: + for link in link_header_list: + url = link[link.index('<')+ 1:link.index('>')] + original_rel = link[link.index('rel="')+ 5:link.rindex('"')] + if (original_rel.lower() == 'describedby'): + rel = 'meta' + else: + rel = original_rel + + if rel in ['meta', 'acl']: + link_dict[rel] = url_join(url, item_url) + + return link_dict + +def parse_link_header_to_array(link_header: str) -> List: + if (not link_header): return + linkexp = '<[^>]*>\s*(\s*;\s*[^()<>@,;:"/[\]?={} \t]+=(([^\(\)<>@,;:"\/\[\]\?={} \t]+)|("[^"]*")))*(,|$)' + match = re.finditer(linkexp, link_header) + links = [x.group() for x in match] + return links + +def url_join(given, base) -> str: + base = str(base) + base_hash = base.find('#') + if (base_hash > 0): + base = base[0:base_hash] + + if (len(given) == 0): + return base + + if (given.find('#') == 0): + return base + given + + colon = given.find(':') + if (colon >= 0) : + return given + + base_colon = base.find(':') + if (len(base) == 0) : + return given + + if (base_colon < 0) : + return given + + if (+ base_colon + 1): + end_index = +base_colon + 1 + else: + end_index = 9e9 + + base_scheme = base[:end_index] + if (given.find('//') == 0) : + return base_scheme + given + + if (base.find('//', base_colon) == base_colon + 1): + base_single = base.find('/', base_colon + 3) + if (base_single < 0): + if (len(base) - base_colon - 3 > 0): + return base + '/' + given + else: + return base_scheme + given + + else: + base_single = base.find('/', base_colon + 1) + if (base_single < 0): + if (len(base) - base_colon - 1 > 0) : + return base + '/' + given + else: + return base_scheme + given + + if (given.find('/') == 0) : + return base[:base_single] + given + + path = base[base_single:] + try: + last_slash = path.rindex('/') + except: + return base_scheme + given + + if (last_slash >= 0 and last_slash < len(path) - 1) : + if (+last_slash + 1): + end_index = +last_slash + 1 + else: + end_index = 9e9 + path = path[:end_index] + + path += given + while (re.match("[^\/]*\/\.\.\/", path)) : + path = re.sub("[^\/]*\/\.\.\/", '', path, 1) + + path = re.sub("\.\/", '', path) + path = re.sub("\/\.$", '/', path) + return base[:base_single] + path + diff --git a/tests/integration_test_solid_api.py b/tests/integration_test_solid_api.py index 0f074d4..b545181 100644 --- a/tests/integration_test_solid_api.py +++ b/tests/integration_test_solid_api.py @@ -167,6 +167,6 @@ def test_file(): lines = resp.text.split('\n') assert lines[4] == '<> dct:title "This is a test file"; contact:personalTitle "Dr.".' - - - + # get item links + links = api.get_item_links(patchedUrl) + assert links == {'acl': f'{patchedUrl}.acl', 'meta': f'{patchedUrl}.meta'} \ No newline at end of file