Coverage for portality / lib / es_data_mapping.py: 97%

1# -*- coding: UTF-8 -*- 

2""" Create mappings from models """ 

3 

4# ~~ESMappings:Library->Elasticsearch:Technology~~ 

5# ~~->Seamless:Library~~ 

6# ~~->DataObj:Library~~ 

7 

8from portality.lib import plugin 

9 

10 

11def get_mappings(app): 

12 """Get the full set of mappings required for the app""" 

13 

14 # LEGACY DEFAULT MAPPINGS 

15 mappings = app.config["MAPPINGS"] 

16 

17 # Ensure legacy mappings also respect any DEFAULT_INDEX_SETTINGS override (e.g. from test.cfg), 

18 # rather than using the hardcoded dict reference set at module import time in settings.py. 

19 default_settings = app.config.get("DEFAULT_INDEX_SETTINGS", {}) 

20 for key in list(mappings.keys()): 

21 if 'settings' in mappings[key]: 

22 mappings[key]['settings'] = default_settings 

23 

24 # TYPE SPECIFIC MAPPINGS 

25 # get the list of classes which carry the type-specific mappings to be loaded 

26 mapping_daos = app.config.get("ELASTIC_SEARCH_MAPPINGS", []) 

27 

28 # load each class and execute the "mappings" function to get the mappings that need to be imported 

29 for cname in mapping_daos: 

30 klazz = plugin.load_class_raw(cname) 

31 mappings[klazz.__type__] = {'mappings': klazz().mappings()} 

32 mappings[klazz.__type__]['settings'] = app.config["DEFAULT_INDEX_SETTINGS"] 

33 

34 return mappings 

35 

36 

37def apply_mapping_opts(field_name, path, spec, mapping_opts): 

38 dot_path = '.'.join(path + (field_name,)) 

39 if dot_path in mapping_opts.get('exceptions', {}): 

40 return mapping_opts['exceptions'][dot_path] 

41 elif spec['coerce'] in mapping_opts['coerces']: 

42 return mapping_opts['coerces'][spec['coerce']] 

43 else: 

44 # We have found a data type in the struct we don't have a map for to ES type. 

45 raise Exception("Mapping error - no mapping found for {}".format(spec['coerce'])) 

46 

47 

48def create_mapping(struct, mapping_opts, path=()): 

49 result = {"properties": {}} 

50 

51 for field, spec in struct.get("fields", {}).items(): 

52 result["properties"][field] = apply_mapping_opts(field, path, spec, mapping_opts) 

53 

54 for field, spec in struct.get("lists", {}).items(): 

55 if "coerce" in spec: 

56 result["properties"][field] = apply_mapping_opts(field, path, spec, mapping_opts) 

57 

58 for struct_name, struct_body in struct.get("structs", {}).items(): 

59 result["properties"][struct_name] = create_mapping(struct_body, mapping_opts, path + (struct_name,)) 

60 

61 dotpath = '.'.join(path) 

62 for field, additional_mapping in mapping_opts.get("additional_mappings", {}).items(): 

63 if field.startswith(dotpath) and "." not in field[len(dotpath) + 1:]: 

64 result["properties"][field[len(dotpath):]] = additional_mapping 

65 

66 return result