Skip to content

API Reference

This page provides a detailed API reference for the gdrive-suite library, automatically generated from the source code doc-strings.


Main Client

The primary interface for interacting with Google Drive and Sheets.

gdrive_suite.GDriveClient

Provides functionality to: - Download files from Google Drive with or without conversion - Upload files to Google Drive - Retrieve Google Sheets data

Source code in src/gdrive_suite/drive/gdrive_client.py
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
class GDriveClient:
    """
    Provides functionality to:
    - Download files from Google Drive with or without conversion
    - Upload files to Google Drive
    - Retrieve Google Sheets data
    """

    def __init__(self, drive_config_manager: GDriveClientConfig):
        """Initializes the GdriveService
        :param drive_config_manager: A ConfigManager that provides
         Outh2 credentials.
        """
        try:
            self._credentials: Optional[Credentials] = (
                drive_config_manager.get_credentials()
            )
            self.drive_service: Any = build(
                "drive", version="v3", credentials=self._credentials
            )
            self._sheets_service: Any = build(
                serviceName="sheets", version="v4", credentials=self._credentials
            )

        except RefreshError as e:
            raise GDriveAuthError(
                f"Failed to refres credentials. please re-authenticate. Details: {e}"
            ) from e

        except Exception as e:
            raise RuntimeError(f"Failed to initialize Google API services: {e}") from e

    def download_file(self, target: DownloadTarget) -> None:
        """Download a file from Google Drive.

        Args:
            target: A `DownloadTarget` object specifyng what a download and
                where to save it.
        Raises:
            APIError: If the download fails due to a Google API error.
            IOError: For local file system errors.
        """
        if target.destination_path.parent:
            target.destination_path.parent.mkdir(parents=True, exist_ok=True)

        try:
            request = (
                self.drive_service.files().export_media(
                    fileId=target.file_id, mimeType=target.mime_type
                )
                if target.mime_type
                else self.drive_service.files().get_media(fileId=target.file_id)
            )
            with open(target.destination_path, "wb") as file_writer:
                downloader: MediaIoBaseDownload = MediaIoBaseDownload(
                    fd=file_writer, request=request
                )
                _execute_download(downloader)

        except HttpError as e:
            raise APIError(
                f"Failed to download file '{target.destination_path.name}' (ID: {target.file_id})."
                f"Reason: {e.resp.status} {e.resp.reason}"
            ) from e
        except IOError as e:
            raise IOError(
                f"An expected error ocurred while downloading '{target.destination_path.name}': {e}"
            ) from e

    def retrieve_file_content(self, file_id: str) -> io.BytesIO:
        """Retrieve a file from Google Drive as BytesIO object
        Args
         file_id: Google ID of the file to get retrieve_file_content.

         Return:
         BytesIO object containing the file content

         Raises:
           APIError: If the retrieve fails due to Google API error.
           IOError: For local file system errors.
        """
        try:
            request = self.drive_service.files().get_media(fileId=file_id)
            file_content: io.BytesIO = io.BytesIO()
            downloader: MediaIoBaseDownload = MediaIoBaseDownload(
                fd=file_content, request=request
            )

            _execute_download(downloader)
            file_content.seek(0)
            return file_content

        except HttpError as e:
            raise APIError(
                f"Failed to retrieve file content for '{file_id}'. "
                f"Reason: {e.resp.status} {e.resp.reason}"
            ) from e
        except IOError as e:
            raise IOError(
                f"An unexpected error ocurred while retrieving content "
                f"for file ID: '{file_id}': {e}"
            )

    def retrieve_sheet_data(
        self, spreadsheet_id: str, sheet_range: str
    ) -> List[List[Any]]:
        """Read data from a Google sheet
        Args:
             spreadsheet_id: ID for the Google sheet
             sheet_range: Range of cells to read (e.j., A1: Z90)

        Returns:
            List of rows containing cell values

        Raises:
            APIError: If retrieving data fails due to a Google API error.
        """
        try:
            result = (
                self._sheets_service.spreadsheets()
                .values()
                .get(spreadsheetId=spreadsheet_id, range=sheet_range)
                .execute()
            )
            return result.get("values", [])

        except HttpError as e:
            raise APIError(
                f"Failed to retrieve data from sheet '{spreadsheet_id}'. "
                f"Reason: {e.resp.status} {e.resp.reason}"
            ) from e

    def upload_file(self, file_path: Path, folder_id: str, **metadata: Any) -> str:
        """Upload file to Google Drive

        Args:
             file_path: Path to the file to the upload
             folder_id: ID of the folder to upload to
             metadata: Additional metadata to add the file

        Raises:
            APIError: If upload fails due to Google API error.
            IOError: For local file system errors.
        """
        if not file_path.is_file():
            raise FileNotFoundError(f"The specific file does not exist: {file_path}")

        try:
            file_metadata = {"name": file_path.name, "parents": [folder_id], **metadata}
            media: MediaFileUpload = MediaFileUpload(str(file_path), resumable=True)
            response = (
                self.drive_service.files()
                .create(body=file_metadata, media_body=media, fields="id")
                .execute()
            )
            return response.get("id")
        except HttpError as e:
            raise APIError(
                f"Failed to upload file {file_path.name}. "
                f"Reason: {e.resp.status} {e.reason}"
            ) from e
        except IOError as e:
            raise IOError(
                f"An unexpected error ocurred during the upload "
                f"of '{file_path.name}': {e}"
            ) from e

    def list_files(self, query: str, **list_params: Any) -> List[Dict[str, Any]]:
        """Helper function to list files in Google Drive
        Args:
         query: A query for filtering results. See "Search for files"
         guide for support syntax.
         list_params: Additional parameters for listing files.
         Return:
         List of searches files.

         Raises:
           APIError: If the list files fails due to Google API error.
        """
        all_files: List[Dict[str, Any]] = []
        page_token: Optional[str] = None
        try:
            while True:
                response: Any = (
                    self.drive_service.files()
                    .list(
                        spaces="drive",
                        q=query,
                        pageToken=page_token,
                        corpora="user",
                        fields="nextPageToken, files(id, name, parents, mime)",
                        **list_params,
                    )
                    .execute()
                )
                all_files.extend(response.get("files", []))
                page_token = response.get("nextPageToken", None)

                if page_token is None:
                    break
            return all_files
        except HttpError as e:
            raise APIError(
                f"Failed to list files with query '{query}'. "
                f"HTTP Error: {e.resp.status} {e.resp.reason}"
            )

    def find_folder_id_by_path(
        self, start_folder_id: str, path_segments: List[str]
    ) -> Optional[str]:
        """
        Find a folder's ID by navigating a path of folder names.
        Args:
            start_folder_id: The ID of the folder to start from.
            path_segments: A list of folder names to navigate.
        Return:
          The ID of the folder.
        """
        current_folder_id: str = start_folder_id
        for segment in path_segments:
            query: str = f"'{current_folder_id}' in parents and "
            "mimeType='application/vnd.google-apps.folder' and "
            f"name='{segment}' and trashed=false"
            items = self.list_files(query)
            if not items:
                logger.opt(colors=True).error(
                    f"Could not find folder '{segment}' within "
                    f"parent ID '{current_folder_id}'"
                )
                return None
            current_folder_id = items[0]["id"]
            logger.opt(colors=True).info(
                f"Found folder segment '{{}}' with ID: '{current_folder_id}'"
            )

        return current_folder_id

__init__(drive_config_manager)

Initializes the GdriveService :param drive_config_manager: A ConfigManager that provides Outh2 credentials.

Source code in src/gdrive_suite/drive/gdrive_client.py
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(self, drive_config_manager: GDriveClientConfig):
    """Initializes the GdriveService
    :param drive_config_manager: A ConfigManager that provides
     Outh2 credentials.
    """
    try:
        self._credentials: Optional[Credentials] = (
            drive_config_manager.get_credentials()
        )
        self.drive_service: Any = build(
            "drive", version="v3", credentials=self._credentials
        )
        self._sheets_service: Any = build(
            serviceName="sheets", version="v4", credentials=self._credentials
        )

    except RefreshError as e:
        raise GDriveAuthError(
            f"Failed to refres credentials. please re-authenticate. Details: {e}"
        ) from e

    except Exception as e:
        raise RuntimeError(f"Failed to initialize Google API services: {e}") from e

download_file(target)

Download a file from Google Drive.

Parameters:

Name Type Description Default
target DownloadTarget

A DownloadTarget object specifyng what a download and where to save it.

required

Raises: APIError: If the download fails due to a Google API error. IOError: For local file system errors.

Source code in src/gdrive_suite/drive/gdrive_client.py
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def download_file(self, target: DownloadTarget) -> None:
    """Download a file from Google Drive.

    Args:
        target: A `DownloadTarget` object specifyng what a download and
            where to save it.
    Raises:
        APIError: If the download fails due to a Google API error.
        IOError: For local file system errors.
    """
    if target.destination_path.parent:
        target.destination_path.parent.mkdir(parents=True, exist_ok=True)

    try:
        request = (
            self.drive_service.files().export_media(
                fileId=target.file_id, mimeType=target.mime_type
            )
            if target.mime_type
            else self.drive_service.files().get_media(fileId=target.file_id)
        )
        with open(target.destination_path, "wb") as file_writer:
            downloader: MediaIoBaseDownload = MediaIoBaseDownload(
                fd=file_writer, request=request
            )
            _execute_download(downloader)

    except HttpError as e:
        raise APIError(
            f"Failed to download file '{target.destination_path.name}' (ID: {target.file_id})."
            f"Reason: {e.resp.status} {e.resp.reason}"
        ) from e
    except IOError as e:
        raise IOError(
            f"An expected error ocurred while downloading '{target.destination_path.name}': {e}"
        ) from e

find_folder_id_by_path(start_folder_id, path_segments)

Find a folder's ID by navigating a path of folder names. Args: start_folder_id: The ID of the folder to start from. path_segments: A list of folder names to navigate. Return: The ID of the folder.

Source code in src/gdrive_suite/drive/gdrive_client.py
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
def find_folder_id_by_path(
    self, start_folder_id: str, path_segments: List[str]
) -> Optional[str]:
    """
    Find a folder's ID by navigating a path of folder names.
    Args:
        start_folder_id: The ID of the folder to start from.
        path_segments: A list of folder names to navigate.
    Return:
      The ID of the folder.
    """
    current_folder_id: str = start_folder_id
    for segment in path_segments:
        query: str = f"'{current_folder_id}' in parents and "
        "mimeType='application/vnd.google-apps.folder' and "
        f"name='{segment}' and trashed=false"
        items = self.list_files(query)
        if not items:
            logger.opt(colors=True).error(
                f"Could not find folder '{segment}' within "
                f"parent ID '{current_folder_id}'"
            )
            return None
        current_folder_id = items[0]["id"]
        logger.opt(colors=True).info(
            f"Found folder segment '{{}}' with ID: '{current_folder_id}'"
        )

    return current_folder_id

list_files(query, **list_params)

Helper function to list files in Google Drive Args: query: A query for filtering results. See "Search for files" guide for support syntax. list_params: Additional parameters for listing files. Return: List of searches files.

Raises: APIError: If the list files fails due to Google API error.

Source code in src/gdrive_suite/drive/gdrive_client.py
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def list_files(self, query: str, **list_params: Any) -> List[Dict[str, Any]]:
    """Helper function to list files in Google Drive
    Args:
     query: A query for filtering results. See "Search for files"
     guide for support syntax.
     list_params: Additional parameters for listing files.
     Return:
     List of searches files.

     Raises:
       APIError: If the list files fails due to Google API error.
    """
    all_files: List[Dict[str, Any]] = []
    page_token: Optional[str] = None
    try:
        while True:
            response: Any = (
                self.drive_service.files()
                .list(
                    spaces="drive",
                    q=query,
                    pageToken=page_token,
                    corpora="user",
                    fields="nextPageToken, files(id, name, parents, mime)",
                    **list_params,
                )
                .execute()
            )
            all_files.extend(response.get("files", []))
            page_token = response.get("nextPageToken", None)

            if page_token is None:
                break
        return all_files
    except HttpError as e:
        raise APIError(
            f"Failed to list files with query '{query}'. "
            f"HTTP Error: {e.resp.status} {e.resp.reason}"
        )

retrieve_file_content(file_id)

Retrieve a file from Google Drive as BytesIO object Args file_id: Google ID of the file to get retrieve_file_content.

Return: BytesIO object containing the file content

Raises: APIError: If the retrieve fails due to Google API error. IOError: For local file system errors.

Source code in src/gdrive_suite/drive/gdrive_client.py
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def retrieve_file_content(self, file_id: str) -> io.BytesIO:
    """Retrieve a file from Google Drive as BytesIO object
    Args
     file_id: Google ID of the file to get retrieve_file_content.

     Return:
     BytesIO object containing the file content

     Raises:
       APIError: If the retrieve fails due to Google API error.
       IOError: For local file system errors.
    """
    try:
        request = self.drive_service.files().get_media(fileId=file_id)
        file_content: io.BytesIO = io.BytesIO()
        downloader: MediaIoBaseDownload = MediaIoBaseDownload(
            fd=file_content, request=request
        )

        _execute_download(downloader)
        file_content.seek(0)
        return file_content

    except HttpError as e:
        raise APIError(
            f"Failed to retrieve file content for '{file_id}'. "
            f"Reason: {e.resp.status} {e.resp.reason}"
        ) from e
    except IOError as e:
        raise IOError(
            f"An unexpected error ocurred while retrieving content "
            f"for file ID: '{file_id}': {e}"
        )

retrieve_sheet_data(spreadsheet_id, sheet_range)

Read data from a Google sheet Args: spreadsheet_id: ID for the Google sheet sheet_range: Range of cells to read (e.j., A1: Z90)

Returns:

Type Description
List[List[Any]]

List of rows containing cell values

Raises:

Type Description
APIError

If retrieving data fails due to a Google API error.

Source code in src/gdrive_suite/drive/gdrive_client.py
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def retrieve_sheet_data(
    self, spreadsheet_id: str, sheet_range: str
) -> List[List[Any]]:
    """Read data from a Google sheet
    Args:
         spreadsheet_id: ID for the Google sheet
         sheet_range: Range of cells to read (e.j., A1: Z90)

    Returns:
        List of rows containing cell values

    Raises:
        APIError: If retrieving data fails due to a Google API error.
    """
    try:
        result = (
            self._sheets_service.spreadsheets()
            .values()
            .get(spreadsheetId=spreadsheet_id, range=sheet_range)
            .execute()
        )
        return result.get("values", [])

    except HttpError as e:
        raise APIError(
            f"Failed to retrieve data from sheet '{spreadsheet_id}'. "
            f"Reason: {e.resp.status} {e.resp.reason}"
        ) from e

upload_file(file_path, folder_id, **metadata)

Upload file to Google Drive

Parameters:

Name Type Description Default
file_path Path

Path to the file to the upload

required
folder_id str

ID of the folder to upload to

required
metadata Any

Additional metadata to add the file

{}

Raises:

Type Description
APIError

If upload fails due to Google API error.

IOError

For local file system errors.

Source code in src/gdrive_suite/drive/gdrive_client.py
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def upload_file(self, file_path: Path, folder_id: str, **metadata: Any) -> str:
    """Upload file to Google Drive

    Args:
         file_path: Path to the file to the upload
         folder_id: ID of the folder to upload to
         metadata: Additional metadata to add the file

    Raises:
        APIError: If upload fails due to Google API error.
        IOError: For local file system errors.
    """
    if not file_path.is_file():
        raise FileNotFoundError(f"The specific file does not exist: {file_path}")

    try:
        file_metadata = {"name": file_path.name, "parents": [folder_id], **metadata}
        media: MediaFileUpload = MediaFileUpload(str(file_path), resumable=True)
        response = (
            self.drive_service.files()
            .create(body=file_metadata, media_body=media, fields="id")
            .execute()
        )
        return response.get("id")
    except HttpError as e:
        raise APIError(
            f"Failed to upload file {file_path.name}. "
            f"Reason: {e.resp.status} {e.reason}"
        ) from e
    except IOError as e:
        raise IOError(
            f"An unexpected error ocurred during the upload "
            f"of '{file_path.name}': {e}"
        ) from e

Configuration

Classes responsible for managing authentication and configuration.

gdrive_suite.GDriveClientConfig

Handles credential management including retrieving, refreshing, and storing oauth2 credentials for Google Drive API access, both local file based and default application credentials found in a server environment.

Source code in src/gdrive_suite/drive/gdrive_client_config.py
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
class GDriveClientConfig:
    """
    Handles credential management including retrieving, refreshing, and
    storing oauth2 credentials for Google Drive API access, both local file
    based and default application credentials found in a server environment.
    """

    def __init__(self, scopes: List[str], gdrive_settings: GDriveSettings):
        """Constructor for the GoogleDriveClientConfig

        Args:
            scopes: A list of OAuth2 scopers required for the application.
            gdrive_settings: The configuration parameters for the Google Drive client.
        Raises:
            ConfigDirectoryError: If config_dir_path is not a valid directory.
        """
        self.scopes: List[str] = scopes
        self.token_file_path: Path = (
            gdrive_settings.config_dir_path / gdrive_settings.token_file_name
        )
        self.credential_file_path: Path = (
            gdrive_settings.config_dir_path / gdrive_settings.credentials_file_name
        )

    def get_credentials(self) -> Optional[Credentials]:
        """
        Retrieves valid Google API credetials using a prioritized strategy.

        Returns:
            A valid 'google.oauth2.credentials.Credentials' object.

        Raises:
            CredentialsNotFoundError: If no credentials can be found or
                generated through any of the available methods.
        """
        creds: Optional[Credentials] = self._get_default_credentials()
        if creds:
            return creds

        logger.opt(colors=True).info(
            "No default credentials. Attempting to load credentials from local files."
        )

        creds = self._get_local_credentials()
        if creds:
            return creds

        raise CredentialsNotFoundError(
            "Failed to obtain credentials. For local execution, ensure "
            f"'{self.credential_file_path.name}' is in '{self.credential_file_path.parent}.'"
            "For server execution, ensure Application Default Credentials are configured."
        )

    def _get_local_credentials(self) -> Optional[Credentials]:
        """
        Manages the entire lifecycle of local, file-baed credentials.

        It attempts to load, validate, and refresh and existing token.
        If that fails, it initiates a new OAuth2 flow.
        Load credentials from the file toke if it exists.
        Returns:
            Credentials object if a token file exists, None otherwise
        """
        creds = self._load_local_token()

        if creds:
            if creds.valid:
                logger.opt(colors=True).success(
                    "Successfully loaded a valid local credentials."
                )
                return creds
            if creds.expired and creds.refresh_token:
                refreshed_creds = self._refresh_credentials(creds)
                if refreshed_creds:
                    self._save_local_token(refreshed_creds)
                    return refreshed_creds

        logger.opt(colors=True).warning(
            "No valid token found. Starting new OAuth2 flow."
        )
        new_creds = self._run_oauth_flow()
        if new_creds:
            self._save_local_token(new_creds)
            return new_creds

        return None

    def _get_default_credentials(self) -> Optional[Credentials]:
        """
        Tries to get credentials from the application default environment.
        This is the method used in server environments like Cloud functions.
        """
        try:
            creds, _ = google.auth.default(scopes=self.scopes)
            if isinstance(creds, Credentials):
                logger.opt(colors=True).success(
                    "Using Application Default Credentials."
                )
                return creds
        except DefaultCredentialsError:
            return None
        return None

    def _load_local_token(self) -> Optional[Credentials]:
        """
        Manages the entire lifecycle of local, file-based credentials.

        It attempts to load, validate, and refresh and existing token.
        If that fails, it initiates a new OAuth2 flow.
        Load credentials from the file toke if it exists.
        Returns:
            Credentials object if a token file exists, None otherwise
        """

        if not self.token_file_path.exists():
            return None
        try:
            return Credentials.from_authorized_user_file(
                str(self.token_file_path), self.scopes
            )
        except Exception as err:
            logger.opt(colors=True).warning(
                f"Could not load token from '{self.token_file_path}': {err}"
            )
            return None

    def _save_local_token(self, creds: Credentials) -> None:
        """Save credentials to the token file with secure permissions."""
        try:
            logger.opt(colors=True).info(
                f"Saving credentials to '{self.token_file_path}'"
            )
            self.token_file_path.parent.mkdir(parents=True, exist_ok=True)
            self.token_file_path.write_text(creds.to_json())
            self.token_file_path.chmod(0o600)
        except (OSError, IOError) as err:
            raise GDriveAuthError(f"Could not write token file {err}")

    def _refresh_credentials(self, creds: Credentials) -> Optional[Credentials]:
        """Refreshes expired credentials and saves the new token."""
        logger.opt(colors=True).info(
            "Credentials expired. Attempting to refresh token."
        )
        try:
            creds.refresh(Request())
            logger.opt(colors=True).success("Successfully refreshed credentials.")
            return creds
        except RefreshError as e:
            logger.opt(colors=True).error(
                f"Token refresh failed: {e}. A new login is required."
            )
            if self.token_file_path.exists():
                self.token_file_path.unlink()
            return None

    def _run_oauth_flow(self) -> Optional[Credentials]:
        """
        Runs the interactive OAuth2 flow to obtain new credentials.
        """
        if not self.credential_file_path.exists():
            logger.opt(colors=True).warning(
                f"Cannot start OAuth flow: secrets file not found at '{self.credential_file_path}"
            )
            return None

        try:
            flow: InstalledAppFlow = InstalledAppFlow.from_client_secrets_file(
                str(self.credential_file_path), self.scopes
            )
            creds_from_flow: Optional[CredentialsTypes] = flow.run_local_server(port=0)

            if isinstance(creds_from_flow, Credentials):
                creds = cast(Credentials, creds_from_flow)
                logger.opt(colors=True).success(
                    "Successfully obtained new credentials via OAuth2 flow."
                )
                return creds
            else:
                logger.opt(colors=True).error(
                    f"OAuth2 flow returned and unexpected credential type: {type(creds_from_flow)}"
                )
                return None

        except Exception as e:
            logger.opt(colors=True).error(
                f"An error ocurred during the OAuth2 flow: {e}"
            )
            return None

__init__(scopes, gdrive_settings)

Constructor for the GoogleDriveClientConfig

Parameters:

Name Type Description Default
scopes List[str]

A list of OAuth2 scopers required for the application.

required
gdrive_settings GDriveSettings

The configuration parameters for the Google Drive client.

required

Raises: ConfigDirectoryError: If config_dir_path is not a valid directory.

Source code in src/gdrive_suite/drive/gdrive_client_config.py
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def __init__(self, scopes: List[str], gdrive_settings: GDriveSettings):
    """Constructor for the GoogleDriveClientConfig

    Args:
        scopes: A list of OAuth2 scopers required for the application.
        gdrive_settings: The configuration parameters for the Google Drive client.
    Raises:
        ConfigDirectoryError: If config_dir_path is not a valid directory.
    """
    self.scopes: List[str] = scopes
    self.token_file_path: Path = (
        gdrive_settings.config_dir_path / gdrive_settings.token_file_name
    )
    self.credential_file_path: Path = (
        gdrive_settings.config_dir_path / gdrive_settings.credentials_file_name
    )

get_credentials()

Retrieves valid Google API credetials using a prioritized strategy.

Returns:

Type Description
Optional[Credentials]

A valid 'google.oauth2.credentials.Credentials' object.

Raises:

Type Description
CredentialsNotFoundError

If no credentials can be found or generated through any of the available methods.

Source code in src/gdrive_suite/drive/gdrive_client_config.py
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def get_credentials(self) -> Optional[Credentials]:
    """
    Retrieves valid Google API credetials using a prioritized strategy.

    Returns:
        A valid 'google.oauth2.credentials.Credentials' object.

    Raises:
        CredentialsNotFoundError: If no credentials can be found or
            generated through any of the available methods.
    """
    creds: Optional[Credentials] = self._get_default_credentials()
    if creds:
        return creds

    logger.opt(colors=True).info(
        "No default credentials. Attempting to load credentials from local files."
    )

    creds = self._get_local_credentials()
    if creds:
        return creds

    raise CredentialsNotFoundError(
        "Failed to obtain credentials. For local execution, ensure "
        f"'{self.credential_file_path.name}' is in '{self.credential_file_path.parent}.'"
        "For server execution, ensure Application Default Credentials are configured."
    )

gdrive_suite.GDriveSettings dataclass

Configuration parameters for Google Drive client.

Attributes:

Name Type Description
config_dir_path Path

Path to the configuration directory where the token and credentials files are stored.

token_file_name str

The filename for the stored user token.

credentials_file_name str

The filename for the client secret file.

Source code in src/gdrive_suite/context/models.py
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
@dataclass(frozen=True)
class GDriveSettings:
    """
    Configuration parameters for Google Drive client.

    Attributes:
        config_dir_path: Path to the configuration directory where the token and credentials files are stored.
        token_file_name: The filename for the stored user token.
        credentials_file_name: The filename for the client secret file.
    """

    config_dir_path: Path
    token_file_name: str
    credentials_file_name: str

gdrive_suite.DownloadTarget dataclass

Specifies the parameters for a file download operation

Attributes:

Name Type Description
file_id str

The unique ID of the file on Google Drive.

destination_path Path

The full local path(including filename) wjere file will be saved.

mime_type Optional[str]

An optional MIME type to convert a Google Workspace to a different format upon download (e.g., 'application/pdf').

Source code in src/gdrive_suite/context/models.py
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@dataclass(frozen=True)
class DownloadTarget:
    """
    Specifies the parameters for a file download operation

    Attributes:
        file_id: The unique ID of the file on Google Drive.
        destination_path: The full local path(including filename) wjere
            file will be saved.
        mime_type: An optional MIME type to convert a Google Workspace
            to a different format upon download (e.g., 'application/pdf').
    """

    file_id: str
    destination_path: Path
    mime_type: Optional[str]

Exceptions

Custom exceptions raised by the library.

gdrive_suite.GDriveSuiteError

Bases: Exception

Base Exception for all errors rised by the gdrive-suite library.

Source code in src/gdrive_suite/gdrive_exceptions.py
 9
10
11
12
class GDriveSuiteError(Exception):
    """Base Exception for all errors rised by the gdrive-suite library."""

    pass

gdrive_suite.GDriveAuthError

Bases: Exception

Base exception for authentication errors in this module.

Source code in src/gdrive_suite/gdrive_exceptions.py
15
16
17
18
class GDriveAuthError(Exception):
    """Base exception for authentication errors in this module."""

    pass

gdrive_suite.CredentialsNotFoundError

Bases: GDriveAuthError

Raised when no valid credentials can be found or generated.

Source code in src/gdrive_suite/gdrive_exceptions.py
21
22
23
24
class CredentialsNotFoundError(GDriveAuthError):
    """Raised when no valid credentials can be found or generated."""

    pass

gdrive_suite.APIError

Bases: GDriveSuiteError

Raised when an API call to a Google service fails.

This exception abstracts away the underlying 'googleapiclient.http.HttpError' providing a consisteng error type for all API interactions within the suite.

Source code in src/gdrive_suite/gdrive_exceptions.py
27
28
29
30
31
32
33
34
class APIError(GDriveSuiteError):
    """Raised when an API call to a Google service fails.

    This exception abstracts away the underlying 'googleapiclient.http.HttpError'
    providing a consisteng error type for all API interactions within the suite.
    """

    pass