Coverage for lib/datamodel/foreignkey.py: 100%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3 

4# Hermes : Change Data Capture (CDC) tool from any source(s) to any target 

5# Copyright (C) 2024 INSA Strasbourg 

6# 

7# This file is part of Hermes. 

8# 

9# Hermes is free software: you can redistribute it and/or modify 

10# it under the terms of the GNU General Public License as published by 

11# the Free Software Foundation, either version 3 of the License, or 

12# (at your option) any later version. 

13# 

14# Hermes is distributed in the hope that it will be useful, 

15# but WITHOUT ANY WARRANTY; without even the implied warranty of 

16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

17# GNU General Public License for more details. 

18# 

19# You should have received a copy of the GNU General Public License 

20# along with Hermes. If not, see <https://www.gnu.org/licenses/>. 

21 

22from typing import TYPE_CHECKING 

23 

24if TYPE_CHECKING: # pragma: no cover 

25 # Only for type hints, won't import at runtime 

26 from lib.datamodel.dataobject import DataObject 

27 from lib.datamodel.datasource import Datasource 

28 

29 

30class HermesCircularForeignkeysRefsError(Exception): 

31 """Raised when the some circular foreign keys references are found""" 

32 

33 

34class ForeignKey: 

35 """Handle foreign keys, and allow to retrieve foreign objects references""" 

36 

37 def __init__( 

38 self, 

39 from_obj: str, 

40 from_attr: str, 

41 to_obj: str, 

42 to_attr: str, 

43 ): 

44 """Setup a new ForeignKey""" 

45 self._from_obj: str = from_obj 

46 self._from_attr: str = from_attr 

47 self._to_obj: str = to_obj 

48 self._to_attr: str = to_attr 

49 

50 self._repr: str = ( 

51 f"<ForeignKey({self._from_obj}.{self._from_attr}" 

52 f" -> {self._to_obj}.{self._to_attr})>" 

53 ) 

54 self._hash: int = hash(self._repr) 

55 

56 def __repr__(self) -> str: 

57 return self._repr 

58 

59 def __hash__(self) -> int: 

60 return self._hash 

61 

62 def __eq__(self, other: "ForeignKey") -> bool: 

63 return hash(self) == hash(other) 

64 

65 @staticmethod 

66 def checkForCircularForeignKeysRefs( 

67 allfkeys: dict[str, list["ForeignKey"]], 

68 fkeys: list["ForeignKey"], 

69 _alreadyMet: list["ForeignKey"] | None = None, 

70 ): 

71 """Will check recursively for circular references in foreign keys, 

72 and raise HermesCircularForeignkeysRefsError if any is found""" 

73 if _alreadyMet is None: 

74 _alreadyMet = [] 

75 

76 for fkey in fkeys: 

77 if fkey in _alreadyMet: 

78 errmsg = ( 

79 f"Circular foreign keys references found in {_alreadyMet}." 

80 " Unable to continue." 

81 ) 

82 __hermes__.logger.critical(errmsg) 

83 raise HermesCircularForeignkeysRefsError(errmsg) 

84 _alreadyMet.append(fkey) 

85 ForeignKey.checkForCircularForeignKeysRefs( 

86 allfkeys, allfkeys[fkey._to_obj], _alreadyMet 

87 ) 

88 

89 @staticmethod 

90 def fetchParentObjs(ds: "Datasource", obj: "DataObject") -> list["DataObject"]: 

91 """Returns a list of parent objects of specified obj from specified 

92 Datasource ds""" 

93 res: list["DataObject"] = [] 

94 objlist = ds[obj.getType()] 

95 for fkey in objlist.FOREIGNKEYS: 

96 parent = ds[fkey._to_obj].get(getattr(obj, fkey._from_attr)) 

97 if parent is not None: 

98 res.append(parent) 

99 res.extend(ForeignKey.fetchParentObjs(ds, parent)) 

100 return res