Coverage for mindsdb / integrations / handlers / openbb_handler / openbb_handler.py: 0%

1from functools import reduce 

2from openbb_core.app.static.app_factory import create_app 

3 

4from mindsdb.integrations.handlers.openbb_handler.openbb_tables import create_table_class 

5from mindsdb.integrations.libs.api_handler import APIHandler 

6from mindsdb.integrations.libs.response import HandlerStatusResponse as StatusResponse 

7from mindsdb.utilities import log 

8from mindsdb.integrations.handlers.openbb_handler.openbb_tables import OpenBBtable 

9 

10logger = log.getLogger(__name__) 

11 

12 

13class OpenBBHandler(APIHandler): 

14 """A class for handling connections and interactions with the OpenBB Platform. 

15 

16 Attributes: 

17 PAT (str): OpenBB's personal access token. Sign up here: https://my.openbb.co 

18 is_connected (bool): Whether or not the user is connected to their OpenBB account. 

19 

20 """ 

21 

22 def __init__(self, name: str = None, **kwargs): 

23 """Registers all API tables and prepares the handler for an API connection. 

24 

25 Args: 

26 name: (str): The handler name to use 

27 """ 

28 super().__init__(name) 

29 self.PAT = None 

30 

31 args = kwargs.get("connection_data", {}) 

32 if "PAT" in args: 

33 self.PAT = args["PAT"] 

34 

35 self.is_connected = False 

36 

37 # Initialize OpenBB 

38 # pylint: disable=import-outside-toplevel 

39 from openbb.package.__extensions__ import Extensions 

40 self.obb = create_app(Extensions) 

41 

42 for cmd in list(self.obb.coverage.command_model.keys()): 

43 

44 openbb_params = self.obb.coverage.command_model[cmd]["openbb"]["QueryParams"] 

45 openbb_data = self.obb.coverage.command_model[cmd]["openbb"]["Data"] 

46 

47 # Creates the default data retrieval function for the given command 

48 # e.g. obb.equity.price.historical, obb.equity.fa.income 

49 # Note: Even though openbb_params just contains the standard fields that are 

50 # common across vendors users are able to select any of the fields from the vendor 

51 # as well. However, some of them might have no effect on the data if the vendor 

52 # doesn't support it. Regardless, the endpoint won't crash. 

53 table_class = create_table_class( 

54 params_metadata=openbb_params, 

55 response_metadata=openbb_data, 

56 obb_function=reduce(getattr, cmd[1:].split('.'), self.obb), 

57 func_docs=f"https://docs.openbb.co/platform/reference/{cmd[1:].replace('.', '/')}" 

58 ) 

59 self._register_table(cmd.replace('.', '_')[1:], table_class(self)) 

60 

61 # Creates the data retrieval function for each provider 

62 # e.g. obb.equity.price.historical_polygon, obb.equity.price.historical_intrinio 

63 for provider in list(self.obb.coverage.command_model[cmd].keys()): 

64 

65 # Skip the openbb provider since we already created it and it will look like obb.equity.price.historical 

66 if provider == "openbb": 

67 continue 

68 

69 provider_extra_params = self.obb.coverage.command_model[cmd][provider]["QueryParams"] 

70 combined_params = provider_extra_params.copy() # create a copy to avoid modifying the original 

71 combined_params["fields"] = {**openbb_params["fields"], **provider_extra_params["fields"]} # merge the fields 

72 

73 provider_extra_data = self.obb.coverage.command_model[cmd][provider]["Data"] 

74 combined_data = provider_extra_data.copy() # create a copy to avoid modifying the original 

75 combined_data["fields"] = {**openbb_data["fields"], **provider_extra_data["fields"]} # merge the fields 

76 

77 table_class = create_table_class( 

78 params_metadata=combined_params, 

79 response_metadata=combined_data, 

80 obb_function=reduce(getattr, cmd[1:].split('.'), self.obb), 

81 func_docs=f"https://docs.openbb.co/platform/reference/{cmd[1:].replace('.', '/')}", 

82 provider=provider 

83 ) 

84 self._register_table(f"{cmd.replace('.', '_')[1:]}_{provider}", table_class(self)) 

85 

86 obb_table = OpenBBtable(self) 

87 self._register_table("openbb_fetcher", obb_table) 

88 

89 def connect(self) -> bool: 

90 """Connects with OpenBB account through personal access token (PAT). 

91 

92 Returns none. 

93 """ 

94 self.is_connected = False 

95 self.obb.account.login(pat=self.PAT) 

96 

97 # Check if PAT utilized is valid 

98 # if obb.user.profile.active: 

99 self.is_connected = True 

100 return True 

101 

102 def check_connection(self) -> StatusResponse: 

103 """Checks connection to OpenBB accounting by checking the validity of the PAT. 

104 

105 Returns StatusResponse indicating whether or not the handler is connected. 

106 """ 

107 

108 response = StatusResponse(False) 

109 

110 try: 

111 if self.connect(): 

112 response.success = True 

113 

114 except Exception as e: 

115 logger.error(f"Error connecting to OpenBB Platform: {e}!") 

116 response.error_message = e 

117 

118 self.is_connected = response.success 

119 return response