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

1import overpy 

2 

3from mindsdb.integrations.handlers.openstreetmap_handler.openstreetmap_tables import (OpenStreetMapNodeTable, 

4 OpenStreetMapWayTable, OpenStreetMapRelationTable) 

5from mindsdb.integrations.libs.api_handler import APIHandler 

6from mindsdb.integrations.libs.response import ( 

7 HandlerStatusResponse as StatusResponse, 

8 HandlerResponse as Response, 

9) 

10from mindsdb.utilities import log 

11from mindsdb_sql_parser import parse_sql 

12 

13logger = log.getLogger(__name__) 

14 

15 

16class OpenStreetMapHandler(APIHandler): 

17 """The OpenStreetMap handler implementation.""" 

18 

19 def __init__(self, name: str, **kwargs): 

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

21 

22 Args: 

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

24 """ 

25 super().__init__(name) 

26 

27 connection_data = kwargs.get("connection_data", {}) 

28 self.connection_data = connection_data 

29 self.kwargs = kwargs 

30 

31 self.connection = None 

32 self.is_connected = False 

33 

34 nodes_data = OpenStreetMapNodeTable(self) 

35 self._register_table("nodes", nodes_data) 

36 

37 ways_data = OpenStreetMapWayTable(self) 

38 self._register_table("ways", ways_data) 

39 

40 relations_data = OpenStreetMapRelationTable(self) 

41 self._register_table("relations", relations_data) 

42 

43 def connect(self) -> StatusResponse: 

44 """Set up the connection required by the handler. 

45 

46 Returns: 

47 StatusResponse: connection object 

48 """ 

49 if self.is_connected is True: 

50 return self.connection 

51 

52 api_session = overpy.Overpass() 

53 

54 self.connection = api_session 

55 

56 self.is_connected = True 

57 

58 return self.connection 

59 

60 def check_connection(self) -> StatusResponse: 

61 """Check connection to the handler. 

62 

63 Returns: 

64 HandlerStatusResponse 

65 """ 

66 response = StatusResponse(False) 

67 

68 try: 

69 api_session = self.connect() 

70 if api_session is not None: 

71 response.success = True 

72 except Exception as e: 

73 logger.error('Error connecting to OpenStreetMap!') 

74 response.error_message = str(e) 

75 

76 self.is_connected = response.success 

77 

78 return response 

79 

80 def native_query(self, query: str) -> Response: 

81 """Execute a native query on the handler. 

82 

83 Args: 

84 query (str): The query to execute. 

85 

86 Returns: 

87 Response: The response from the query. 

88 """ 

89 ast = parse_sql(query) 

90 return self.query(ast)