Coverage for ckanext/udc/graph/preload.py: 15%

1from graphdb_importer import import_and_wait, set_config 

2import ckan.plugins.toolkit as tk 

3import ckan.plugins as plugins 

4import requests 

5import os 

6from .sparql_client import SparqlClient 

7 

8 

9def preload_ontologies(config, graphdb_endpoint: str, username: str, password: str, sparql_client: SparqlClient): 

10 # Download and import ontologies 

11 storage_path = tk.config.get('ckan.storage_path') or './' 

12 download_path = os.path.join(storage_path, 'preload_ontologies') 

13 os.makedirs(download_path, exist_ok=True) 

14 

15 base_api, repo = graphdb_endpoint.split('/repositories/') 

16 set_config(base_api=base_api, repo=repo, username=username, password=password) 

17 

18 for item in config["preload_ontologies"]: 

19 filename = item["ontology_url"].rsplit('/', 1)[1] 

20 path = os.path.join(download_path, filename) 

21 r = requests.get(item["ontology_url"], allow_redirects=True) 

22 with open(path, 'wb') as f: 

23 f.write(r.content) 

24 

25 # Keep retry for 5 times 

26 for cnt in range(5): 

27 try: 

28 try: 

29 import_and_wait(path, replace_graph=True, named_graph=item["graph"]) 

30 except Exception as e: 

31 if 'already scheduled for import' not in str(e): 

32 raise e 

33 break 

34 except Exception as e: 

35 print(f"Error importing {path}: {e}") 

36 print("Retrying...") 

37 if cnt == 4: 

38 # Restart(Exit) the server if it fails 5 times 

39 print("Exiting after 5 retries") 

40 exit(1) 

41 

42 

43 # Preload options for dropdowns 

44 dropdown_reload(maturity_model=config["maturity_model"]) 

45 

46 

47def dropdown_reload(name=None, maturity_model=None): 

48 """ 

49 Reload dropdown options "optionsFromQuery". 

50 Set `name` to None to reload all dropdown options. 

51 """ 

52 if plugins.get_plugin('udc').disable_graphdb: 

53 return 

54 

55 client = plugins.get_plugin('udc').sparql_client 

56 

57 if maturity_model is None: 

58 maturity_model = plugins.get_plugin('udc').maturity_model 

59 

60 for level in maturity_model: 

61 for field in level["fields"]: 

62 if name is not None and field.get("name") != name: 

63 continue 

64 if field.get("optionsFromQuery"): 

65 options = [] 

66 if field["type"] == "single_select": 

67 options.append({ 

68 "text": "Please select", 

69 "value": "" 

70 }) 

71 textVar = field["optionsFromQuery"]["text"] 

72 valueVar = field["optionsFromQuery"]["value"] 

73 result = client.execute_sparql(field["optionsFromQuery"]["query"]) 

74 for item in result["results"]["bindings"]: 

75 options.append({ 

76 "text": item[textVar]["value"], 

77 "value": item[valueVar]["value"], 

78 }) 

79 field["options"] = options 

80