Coverage for mindsdb / integrations / handlers / google_books_handler / google_books_handler.py: 0%
97 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-21 00:36 +0000
1import os
2from google.auth.transport.requests import Request
3from google.oauth2 import service_account
4from google.oauth2.credentials import Credentials
5from googleapiclient.discovery import build
6import pandas as pd
7from pandas import DataFrame
8from mindsdb.api.executor.data_types.response_type import RESPONSE_TYPE
9from .google_books_tables import BookshelvesTable, VolumesTable
10from mindsdb.integrations.libs.api_handler import APIHandler, FuncParser
11from mindsdb.integrations.libs.response import (
12 HandlerStatusResponse as StatusResponse,
13 HandlerResponse as Response,
14)
15from mindsdb.utilities import log
17logger = log.getLogger(__name__)
20class GoogleBooksHandler(APIHandler):
21 """
22 A class for handling connections and interactions with the Google Books API.
23 """
25 name = "google_books"
27 def __init__(self, name: str, **kwargs):
28 """
29 Initialize the Google Books API handler.
30 Args:
31 name (str): name of the handler
32 kwargs (dict): additional arguments
33 """
34 super().__init__(name)
35 self.token = None
36 self.service = None
37 self.connection_data = kwargs.get("connection_data", {})
38 self.credentials_file = self.connection_data.get("credentials", None)
39 self.credentials = None
40 self.scopes = ["https://www.googleapis.com/auth/books"]
41 self.is_connected = False
42 self.connection = None
43 bookshelves = BookshelvesTable(self)
44 self.bookshelves = bookshelves
45 self._register_table("bookshelves", bookshelves)
46 volumes = VolumesTable(self)
47 self.volumes = volumes
48 self._register_table("volumes", volumes)
50 def connect(self):
51 """
52 Set up any connections required by the handler
53 Should return output of check_connection() method after attempting
54 connection. Should switch self.is_connected.
55 Returns:
56 HandlerStatusResponse
57 """
59 if self.is_connected is True:
60 return self.service
61 if self.credentials_file:
62 if os.path.exists("token_books.json"):
63 self.credentials = Credentials.from_authorized_user_file("token_books.json", self.scopes)
64 if not self.credentials or not self.credentials.valid:
65 if self.credentials and self.credentials.expired and self.credentials.refresh_token:
66 self.credentials.refresh(Request())
67 else:
68 self.credentials = service_account.Credentials.from_service_account_file(
69 self.credentials_file, scopes=self.scopes
70 )
71 # Save the credentials for the next run
72 with open("token_books.json", "w") as token:
73 token.write(self.credentials.to_json())
74 self.service = build("books", "v1", credentials=self.credentials)
75 return self.service
77 def check_connection(self) -> StatusResponse:
78 """
79 Check connection to the handler
80 Returns:
81 HandlerStatusResponse
82 """
84 response = StatusResponse(False)
86 try:
87 self.connect()
88 response.success = True
89 except Exception as e:
90 logger.error(f"Error connecting to Google Books API: {e}!")
91 response.error_message = e
93 self.is_connected = response.success
94 return response
96 def native_query(self, query: str = None) -> Response:
97 """
98 Receive raw query and act upon it somehow.
99 Args:
100 query (Any): query in native format (str for sql databases,
101 api's json etc)
102 Returns:
103 HandlerResponse
104 """
105 method_name, params = FuncParser().from_string(query)
107 df = self.call_application_api(method_name, params)
109 return Response(RESPONSE_TYPE.TABLE, data_frame=df)
111 def get_bookshelves(self, params: dict = None) -> DataFrame:
112 """
113 Get bookshelf from Google Books API
114 Args:
115 params (dict): query parameters
116 Returns:
117 DataFrame
118 """
119 service = self.connect()
120 minShelf = None
121 maxShelf = None
122 if params["shelf"]:
123 shelf = int(params["shelf"])
124 if params["source"]:
125 df = (
126 service.mylibrary()
127 .bookshelves()
128 .get(shelf=shelf, userid=params["userid"], source=params["source"])
129 .execute()
130 )
131 else:
132 df = service.mylibrary().bookshelves().get(shelf=shelf, userid=params["userid"]).execute()
134 df = pd.DataFrame(df, columns=self.bookshelves.get_columns())
135 return df
136 elif not params["minShelf"] and params["maxShelf"]:
137 minShelf = int(params["maxShelf"]) - 10
138 maxShelf = int(params["maxShelf"])
139 elif not params["maxShelf"] and params["minShelf"]:
140 minShelf = int(params["minShelf"])
141 maxShelf = int(params["minShelf"]) + 10
142 elif params["maxShelf"] and params["minShelf"]:
143 minShelf = int(params["minShelf"])
144 maxShelf = int(params["maxShelf"])
146 args = {
147 key: value for key, value in params.items() if key not in ["minShelf", "maxShelf"] and value is not None
148 }
149 bookshelves = service.bookshelves().list(userid=params["userid"], **args).execute()
151 df = pd.DataFrame(bookshelves["items"], columns=self.bookshelves.get_columns())
152 if minShelf is not None or maxShelf is not None:
153 # Drop bookshelves that are not in the id range
154 df = df.drop(df[(df["id"] < minShelf) | (df["id"] > maxShelf)].index)
156 return df
158 def get_volumes(self, params: dict = None) -> DataFrame:
159 """
160 Get volumes from Google Books API
161 Args:
162 params (dict): query parameters
163 Returns:
164 DataFrame
165 """
166 service = self.connect()
167 args = {key: value for key, value in params.items() if value is not None}
168 volumes = service.volumes().list(**args).execute()
169 df = pd.DataFrame(volumes["items"], columns=self.volumes.get_columns())
170 return df
172 def call_application_api(self, method_name: str = None, params: dict = None) -> DataFrame:
173 """
174 Call Google Books API and map the data to pandas DataFrame
175 Args:
176 method_name (str): method name
177 params (dict): query parameters
178 Returns:
179 DataFrame
180 """
181 if method_name == "get_bookshelves":
182 return self.get_bookshelves(params)
183 elif method_name == "get_volumes":
184 return self.get_volumes(params)
185 else:
186 raise NotImplementedError(f"Unknown method {method_name}")