Skip to content

Reference Guides

Document

Bases: DriveObject

A document on google drive.

Source code in pygsuite/docs/document.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 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
class Document(DriveObject):
    """A document on google drive."""

    _mimetype = MimeType.DOCS
    _base_url = "https://docs.google.com/document/d/{}/edit"

    def __init__(self, id=None, client=None, name=None, _document=None, local=False):

        if not local:
            client = client or Clients.docs_client
        self.service = client
        self.id = parse_id(id) if id else None
        DriveObject.__init__(self, id=id, client=client)
        self._document = _document or client.documents().get(documentId=self.id).execute()
        self._change_queue = []
        self.auto_sync = False

    def id(self):
        return self._document["id"]

    def _mutation(self, reqs, flush=False):
        if not reqs:
            return None
        self._change_queue += reqs
        if flush or self.auto_sync:
            return self.flush()

    @retry(HttpError, tries=3, delay=5, backoff=3)
    def flush(self, reverse=False):
        if reverse:
            base = reversed(self._change_queue)
        else:
            base = self._change_queue
        final = []
        for item in base:
            if isinstance(item, list):
                for i in item:
                    final.append(i)
            else:
                final.append(item)
        if not base:
            return []
        out = (
            self.service.documents()
            .batchUpdate(body={"requests": final}, documentId=self.id)
            .execute()["replies"]
        )

        self._change_queue = []
        self.refresh()
        return out

    # def delete(self, start=0, end=None, flush=True):
    #     end = end or self.body.end_index
    #     self._mutation([{'deleteContentRange': {'range': {
    #         "segmentId": None,
    #         "startIndex": start,
    #         "endIndex": end
    #     }}}])
    #     if flush:
    #         self.flush()

    @property
    def body(self):
        return Body(self._document.get("body"), self)

    @property
    def footers(self):
        return [Footers(item, self) for item in self._document.get("footers")]

    @property
    def footnotes(self):
        return [Footnotes(item, self) for item in self._document.get("footnotes")]

    @property
    def headers(self):
        return [Headers(item, self) for item in self._document.get("headers")]

    @property
    def title(self):
        return self._document.get("title")

    @title.setter
    def title(self, title: str):
        raise NotImplementedError

    def refresh(self):
        self._document = self.service.documents().get(documentId=self.id).execute()
        self.body._pending = []

    @property
    def url(self):
        return f"https://docs.google.com/docs/d/{self.id}"

Drive

Due for deprecation, please use File and Folder to search and create objects.

Source code in pygsuite/drive/drive.py 
18
19
20
21
22
23
24
25
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
class Drive:
    """Due for deprecation, please use File and Folder to search and create objects."""

    def __init__(self, client=None):
        client = client or Clients.drive_client_v3
        self.service = client

        # warn users about deprecation
        warnings.warn(
            (
                "This object and its methods will be deprecated soon."
                "Please consider using a drive.File or drive.Folder object instead"
            ),
            DeprecationWarning,
        )

    def _find_files(self, type: MimeType, name: Optional[str] = None):
        q = f'mimeType="{type}"'
        if name:
            q += f' and name = "{name}"'
        base = self.service.files().list(q=q).execute()
        files = base.get("files")
        page_token = base.get("nextPageToken")
        while page_token is not None:
            base = self.service.files().list(q=q, page_token=page_token).execute()
            files += base.get("files")
            page_token = base.get("nextPageToken")
        return files

    def find_files(
        self,
        folder_id: Optional[str] = None,
        name: Optional[str] = None,
        exact_match: bool = True,
        type: Optional[Union[MimeType, str]] = None,
        extra_conditions: Optional[Union[QueryString, QueryStringGroup]] = None,
        support_all_drives: bool = False,
    ):
        """Find matching files given certain criteria.

        Args:
            folder_id (str): The folder ID to search within. If none is provided, a recursive search of all folders is performed.
            name (str): The case-sensitive name of the file to search for.
            exact_match (bool): Whether to only match the given name exactly, or return any name containing the string.
            type (Union[MimeType, str]): A specific Google Docs type to match.
            extra_conditions (Union[QueryString, QueryStringGroup]): Any additional queries to pass to the files search.
            support_all_drives (bool): Whether the requesting application supports both My Drives and shared drives.
        """
        query_components: List[Union[QueryString, QueryStringGroup]] = []

        # name match query
        if name:
            operator = Operator.EQUAL if exact_match else Operator.CONTAINS
            name_query = QueryString(QueryTerm.NAME, operator, name)
            query_components.append(name_query)

        # optional type query
        if type:
            mimetype = str(type) if isinstance(type, MimeType) else type
            type_query = QueryString(QueryTerm.MIMETYPE, Operator.EQUAL, mimetype)
            query_components.append(type_query)

        # optional auxiliary query
        if extra_conditions:
            query_components.append(extra_conditions)

        final_query = QueryStringGroup(query_components) if query_components else None
        folder = Folder(id=folder_id, client=self.service)
        files = folder.get_files(
            extra_conditions=final_query, support_all_drives=support_all_drives
        )

        return files

    def copy_file(self, file_id, title: str, folder_id: str):
        body = {"name": title, "parents": [folder_id]}
        response = self.service.files().copy(fileId=file_id, body=body).execute()
        return response

find_files(folder_id=None, name=None, exact_match=True, type=None, extra_conditions=None, support_all_drives=False)

Find matching files given certain criteria.

Parameters:

Name Type Description Default
folder_id str

The folder ID to search within. If none is provided, a recursive search of all folders is performed.

 None
name str

The case-sensitive name of the file to search for.

 None
exact_match bool

Whether to only match the given name exactly, or return any name containing the string.

 True
type Union[MimeType, str]

A specific Google Docs type to match.

 None
extra_conditions Union[QueryString, QueryStringGroup]

Any additional queries to pass to the files search.

 None
support_all_drives bool

Whether the requesting application supports both My Drives and shared drives.

 False
Source code in pygsuite/drive/drive.py 
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
def find_files(
    self,
    folder_id: Optional[str] = None,
    name: Optional[str] = None,
    exact_match: bool = True,
    type: Optional[Union[MimeType, str]] = None,
    extra_conditions: Optional[Union[QueryString, QueryStringGroup]] = None,
    support_all_drives: bool = False,
):
    """Find matching files given certain criteria.

    Args:
        folder_id (str): The folder ID to search within. If none is provided, a recursive search of all folders is performed.
        name (str): The case-sensitive name of the file to search for.
        exact_match (bool): Whether to only match the given name exactly, or return any name containing the string.
        type (Union[MimeType, str]): A specific Google Docs type to match.
        extra_conditions (Union[QueryString, QueryStringGroup]): Any additional queries to pass to the files search.
        support_all_drives (bool): Whether the requesting application supports both My Drives and shared drives.
    """
    query_components: List[Union[QueryString, QueryStringGroup]] = []

    # name match query
    if name:
        operator = Operator.EQUAL if exact_match else Operator.CONTAINS
        name_query = QueryString(QueryTerm.NAME, operator, name)
        query_components.append(name_query)

    # optional type query
    if type:
        mimetype = str(type) if isinstance(type, MimeType) else type
        type_query = QueryString(QueryTerm.MIMETYPE, Operator.EQUAL, mimetype)
        query_components.append(type_query)

    # optional auxiliary query
    if extra_conditions:
        query_components.append(extra_conditions)

    final_query = QueryStringGroup(query_components) if query_components else None
    folder = Folder(id=folder_id, client=self.service)
    files = folder.get_files(
        extra_conditions=final_query, support_all_drives=support_all_drives
    )

    return files

File

Bases: DriveObject

Base class for a Google Drive File

Source code in pygsuite/drive/file.py 
10
11
12
13
14
15
16
17
18
19
20
21
class File(DriveObject):
    """Base class for a Google Drive File"""

    _mimetype = MimeType.UNKNOWN
    _base_url = "https://drive.google.com/file/d/{}/view"

    def __init__(self, id: str = None, client: Optional[Resource] = None):

        self.id = parse_id(id) if id else None
        self.client = client

        DriveObject.__init__(self, id=id, client=client)

Folder

Bases: DriveObject

Base class for a Google Drive Folder

Source code in pygsuite/drive/folder.py 
14
15
16
17
18
19
20
21
22
23
24
25
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
class Folder(DriveObject):
    """Base class for a Google Drive Folder"""

    _mimetype = MimeType.FOLDER
    _base_url = "https://drive.google.com/drive/folders/{}"

    def __init__(self, id: str = None, client: Optional[Resource] = None):

        self.id = parse_id(id) if id else None
        self.client: Resource = client or Clients.drive_client

        DriveObject.__init__(self, id=id, client=self.client)

    def get_files(
        self,
        extra_conditions: Optional[Union[QueryString, QueryStringGroup]] = None,
        support_all_drives: bool = True,
    ) -> List[File]:
        """The files in a given folder. If no folder ID is given in the instance,
        a recursive search is performed from the Drive root.

        Args:
            extra_conditions (Union[QueryString, QueryStringGroup]): Any additional queries to pass to the files search.
            support_all_drives (bool): Whether the requesting application supports both My Drives and shared drives.

        Returns a list of any Files found.
        """
        query_components: List[Union[QueryString, QueryStringGroup]] = []

        if self.id:
            query_components.append(QueryString(QueryTerm.PARENTS, Operator.IN, self.id))

        if extra_conditions:
            query_components.append(extra_conditions)

        query = QueryStringGroup(query_components).formatted if query_components else None

        files = execute_paginated_command(
            client=self.client.files(),
            method="list",
            fetch_field="files",
            q=query,
            spaces="drive",
            fields="nextPageToken, files(id, name)",
            supportsAllDrives=support_all_drives,
            includeItemsFromAllDrives=support_all_drives,
        )

        return [File(file.get("id")) for file in files]

get_files(extra_conditions=None, support_all_drives=True)

The files in a given folder. If no folder ID is given in the instance, a recursive search is performed from the Drive root.

Parameters:

Name Type Description Default
extra_conditions Union[QueryString, QueryStringGroup]

Any additional queries to pass to the files search.

 None
support_all_drives bool

Whether the requesting application supports both My Drives and shared drives.

 True

Returns a list of any Files found.

Source code in pygsuite/drive/folder.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
def get_files(
    self,
    extra_conditions: Optional[Union[QueryString, QueryStringGroup]] = None,
    support_all_drives: bool = True,
) -> List[File]:
    """The files in a given folder. If no folder ID is given in the instance,
    a recursive search is performed from the Drive root.

    Args:
        extra_conditions (Union[QueryString, QueryStringGroup]): Any additional queries to pass to the files search.
        support_all_drives (bool): Whether the requesting application supports both My Drives and shared drives.

    Returns a list of any Files found.
    """
    query_components: List[Union[QueryString, QueryStringGroup]] = []

    if self.id:
        query_components.append(QueryString(QueryTerm.PARENTS, Operator.IN, self.id))

    if extra_conditions:
        query_components.append(extra_conditions)

    query = QueryStringGroup(query_components).formatted if query_components else None

    files = execute_paginated_command(
        client=self.client.files(),
        method="list",
        fetch_field="files",
        q=query,
        spaces="drive",
        fields="nextPageToken, files(id, name)",
        supportsAllDrives=support_all_drives,
        includeItemsFromAllDrives=support_all_drives,
    )

    return [File(file.get("id")) for file in files]

Form

Bases: BaseForm, DriveObject

A form on google drive.

Source code in pygsuite/forms/form.py 
 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
class Form(BaseForm, DriveObject):
    """A form on google drive."""

    _mimetype = MimeType.FORMS

    _base_url = "https://docs.google.com/forms/d/{}/edit"

    def __init__(self, id: str = None, client=None, name=None, _form=None, local: bool = False):

        if not local:
            client = client or Clients.forms_client
        self.service = client
        self.id = parse_id(id) if id else None
        DriveObject.__init__(self, id=self.id, client=client)
        BaseForm.__init__(self, object_info=_form or client.forms().get(formId=self.id).execute())
        self._change_queue: List = []
        self.auto_sync: bool = False
        self._info_cache: Optional["Info"] = None
        self._settings_cache: Optional["FormSettings"] = None
        self._items_cache: Optional[List["Item"]] = None

        def update_factory(idx, item):
            self._mutation([UpdateItemRequest(item=item, location=Location(index=idx)).wire_format])

        self.items_update_factory = update_factory

        def delete_factory(idx):
            self._mutation([DeleteItemRequest(location=Location(index=idx)).wire_format])

        self.items_delete_factory = delete_factory

        def move_factory(original_idx, new_idx):
            self._mutation(
                [
                    MoveItemRequest(
                        new_location=Location(new_idx), original_location=Location(original_idx)
                    ).wire_format
                ]
            )

        self.items_move_factory = move_factory

        def create_factory(item, idx):
            self._mutation([CreateItemRequest(item, location=Location(idx)).wire_format])

        self.items_create_factory = create_factory

    def _mutation(self, reqs, flush: bool = False):
        if not reqs:
            return None
        self._change_queue += reqs
        if flush or self.auto_sync:
            return self.flush()

    @retry(HttpError, tries=3, delay=5, backoff=3, fatal_exceptions=(FatalHttpError,))
    def batch_update(self, requests):
        try:
            return (
                self.service.forms()
                .batchUpdate(body={"requests": requests}, formId=self.id)
                .execute()["replies"]
            )
        except HttpError as e:
            if e.status_code in FATAL_HTTP_CODES:
                raise FatalHttpError(e.resp, e.content, e.uri)
            raise e

    def flush(self, reverse=False):
        if reverse:
            base = reversed(self._change_queue)
        else:
            base = self._change_queue
        final = []
        for item in base:
            if isinstance(item, list):
                for i in item:
                    final.append(i)
            else:
                final.append(item)
        for z in final:
            logger.debug(z)
        if not base:
            return []
        out = self.batch_update(final)
        self._change_queue = []
        self.refresh()
        return out

    @property
    def info(self) -> "Info":
        if self._info_cache:
            return self._info_cache
        item = super().info
        # manually set update mask here
        if isinstance(item._info, WatchedDictionary):
            return item
        uf = lambda: self._mutation(  # noqa: E731
            [UpdateFormInfoRequest(info=item, update_mask="*").wire_format]
        )
        item._info = WatchedDictionary(parent_dict=item._info, update_factory=uf)
        self._info_cache = item
        return item

    @info.setter
    def info(self, item: "Info"):
        uf = lambda: self._mutation(  # noqa: E731
            [UpdateFormInfoRequest(info=item, update_mask="*").wire_format]
        )

        item._info = WatchedDictionary(
            parent_dict=item._info, update_factory=uf, default_flush=True
        )
        super(Form, self.__class__).info.fset(self, item)  # type: ignore
        self._info_cache = None

    @property
    def settings(self) -> "FormSettings":
        if self._settings_cache:
            return self._settings_cache
        from pygsuite.forms.generated.update_settings_request import UpdateSettingsRequest

        settings = super().settings
        if isinstance(settings._info, WatchedDictionary):
            return settings
        uf = lambda: self._mutation(  # noqa: E731
            [UpdateSettingsRequest(settings=settings, update_mask="*").wire_format]
        )
        settings._info = WatchedDictionary(parent_dict=settings._info, update_factory=uf)
        self._settings_cache = settings
        return settings

    @settings.setter
    def settings(self, settings: "FormSettings"):
        from pygsuite.forms.generated.update_settings_request import UpdateSettingsRequest

        uf = lambda: self._mutation(  # noqa: E731
            [UpdateSettingsRequest(settings=settings, update_mask="*").wire_format]
        )
        settings._info = WatchedDictionary(
            parent_dict=settings._info, update_factory=uf, default_flush=True
        )
        super(Form, self.__class__).settings.fset(self, settings)  # type:ignore
        self._settings_cache = None

    @property
    def items(self) -> List["Item"]:
        if self._items_cache:
            return self._items_cache
        base = super().items
        if not isinstance(base, WatchedList):
            base = WatchedList(
                iterable=base,
                update_factory=self.items_update_factory,
                delete_factory=self.items_delete_factory,
                move_factory=self.items_move_factory,
                create_factory=self.items_create_factory,
            )
        self._items_cache = base
        return base

    @items.setter
    def items(self, items: List["Item"]):  # type: ignore
        items = WatchedList(
            iterable=items,
            update_factory=self.items_update_factory,
            delete_factory=self.items_delete_factory,
            move_factory=self.items_move_factory,
            create_factory=self.items_create_factory,
        )
        # handle assignments/overwrites
        # someone could directly
        if self._items_cache and items != self._items_cache:
            # it may be possible to optimize this
            # but for now, take the brute force approach
            # on assignment, remove all current items
            # then trigger additions of the new objects
            for idx, val in reversed(list(enumerate(self._items_cache))):
                del self._items_cache[idx]
            # then set back
            for idx, val in enumerate(items):
                self._items_cache.append(val)
        super(Form, self.__class__).items.fset(self, items)  # type: ignore
        self._items_cache = None

    @retry(HttpError, tries=3, delay=5, backoff=3, fatal_exceptions=(FatalHttpError,))
    def refresh(self):
        self._info = self.service.forms().get(formId=self.id).execute()
        self._info_cache = None
        self._settings_cache = None
        self._items_cache = None

Spreadsheet

Bases: DriveObject

Base class for the GSuite Spreadsheets API.

Source code in pygsuite/sheets/sheet.py 
 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
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
class Spreadsheet(DriveObject):
    """Base class for the GSuite Spreadsheets API."""

    _mimetype = MimeType.SHEETS
    _base_url = "https://docs.google.com/spreadsheets/d/{}/edit"

    def __init__(self, id: str, client: Optional[Resource] = None):
        """Method to initialize the class.

        The __init__ method accepts a client connection to the Google API, which it uses to retrieve the properties
        of the spreadsheet and create a _spreadsheet object to execute other requests of the API.

        Args:
            id (str): id of the Google Spreadsheet. Spreadsheet id can be found in the Spreadsheet URL between /d/ and /edit, eg:
                https://docs.google.com/spreadsheets/d/<spreadsheetId>/edit#gid=0
            client (googleapiclient.discovery.Resource): connection to the Google API Sheets resource.
        """

        self.service = client or Clients.sheets_client
        self.id = parse_id(id) if id else None
        DriveObject.__init__(self, id=id, client=client)

        self._spreadsheet = self._execute(self.service.spreadsheets().get(spreadsheetId=self.id))
        self._properties = self._spreadsheet.get("properties")

        # queues to add to and run in flush()
        self._spreadsheets_update_queue: List[Dict] = []
        self._values_update_queue: List[Dict] = []

    def __getitem__(self, key: Union[str, int]):

        # if the key is an integer, use the int as worksheet index
        if isinstance(key, int):
            return self.worksheets[key]
        # if the key is a string, try to match to a worksheet name
        elif isinstance(key, str):
            try:
                return [worksheet for worksheet in self.worksheets if worksheet.name == key][0]
            except IndexError:
                raise KeyError(
                    "No worksheet with the title '{key}' exists. The worksheets included in your spreadsheet are: {worksheets}".format(
                        key=key, worksheets=[worksheet.name for worksheet in self.worksheets]
                    )
                )
        else:
            raise ValueError(
                "The key must be a string or integer. Use a string to get a worksheet by name, and integer to get a worksheet by index."
            )

    @property
    def worksheets(self):
        """List of Worksheet objects for each worksheet in the Spreadsheet."""
        return [Worksheet(sheet, self) for sheet in self._spreadsheet.get("sheets")]

    def refresh(self):
        """Method to refresh the spreadsheets API connection."""
        self._spreadsheet = self._execute(self.service.spreadsheets().get(spreadsheetId=self.id))

        self._spreadsheets_update_queue = []
        self._values_update_queue = []

    @retry((HttpError), tries=3, delay=10, backoff=5)
    def flush(
        self,
        reverse: bool = False,
        value_input_option: ValueInputOption = ValueInputOption.USER_ENTERED,
        include_values_in_response: bool = False,
        response_value_render_option: ValueRenderOption = ValueRenderOption.FORMATTED_VALUE,
        response_date_time_render_option: DateTimeRenderOption = DateTimeRenderOption.SERIAL_NUMBER,
    ):
        """Method to execute the change queues created through other methods.

        Args:
            reverse (bool): If True, the changes queued will be executed in reverse order.
            value_input_option (ValueInputOption): Indicates which dimension an operation should apply to.
            include_values_in_response (bool): If True, the API response will include the values changed.
            response_value_render_option (ValueRenderOption): Determines how values should be rendered in the output.
            response_date_time_render_option (DateTimeRenderOption): Determines how dates should be rendered in the output.

        Returns:
            response_dict (dict): Dictionary of API response JSON.
        """

        if reverse:
            _spreadsheets_update_queue = list(reversed(self._spreadsheets_update_queue))
            _values_update_queue = list(reversed(self._values_update_queue))
        else:
            _spreadsheets_update_queue = self._spreadsheets_update_queue
            _values_update_queue = self._values_update_queue

        response_dict = dict()

        if len(_spreadsheets_update_queue) > 0:
            response_dict["spreadsheets_update_response"] = self._execute(
                self.service.spreadsheets().batchUpdate(
                    body={"requests": _spreadsheets_update_queue}, spreadsheetId=self.id
                )
            ).get("responses")

        if len(_values_update_queue) > 0:
            response_dict["values_update_response"] = self._execute(
                self.service.spreadsheets()
                .values()
                .batchUpdate(
                    spreadsheetId=self.id,
                    body={
                        "valueInputOption": value_input_option.value,
                        "data": _values_update_queue,
                        "includeValuesInResponse": include_values_in_response,
                        "responseValueRenderOption": response_value_render_option.value,
                        "responseDateTimeRenderOption": response_date_time_render_option.value,
                    },
                )
            ).get("responses")

        # the queues are reset in the refresh() method
        # self._spreadsheets_update_queue = []
        # self._values_update_queue = []
        self.refresh()

        return response_dict

    def create_sheet(self, sheet_properties: Optional[SheetProperties] = None, **kwargs):
        """Add a new sheet to the spreadsheet."""
        sheet_properties = sheet_properties or SheetProperties(**kwargs)
        base = {"addSheet": {"properties": sheet_properties.to_json()}}
        self._spreadsheets_update_queue.append(base)
        return self

    def create_and_return_sheet(
        self, sheet_properties: Optional[SheetProperties] = None, **kwargs
    ) -> "Worksheet":
        before = [sheet.name for sheet in self.worksheets]
        self.create_sheet(sheet_properties=sheet_properties, **kwargs)
        self.flush()
        try:
            new = [sheet.name for sheet in self.worksheets if sheet.name not in before][0]
        except IndexError:
            raise ValueError("Error creating new sheet!")
        return self[new]

    def update_sheet(self, worksheet, sheet_properties):

        pass

    def clear_range(self, range: str):
        self._execute(
            self.service.spreadsheets().values().clear(spreadsheetId=self.id, range=range)
        )
        return self

    def delete_sheet(self, title: Optional[str] = None, id: Optional[int] = None):
        """Method to delete a sheet by the title or id.

        Args:
            title (Optional: str): The title of a sheet to delete.
            id (Optional: int): The id of a sheet to delete.

        Returns:
            self (Spreadsheet): The object itself, to method chain.
        """

        # if an id is not given, title must be given; get id from title
        if not id:
            assert title is not None
            id = self[title].id

        base = {"deleteSheet": {"sheetId": id}}
        self._spreadsheets_update_queue.append(base)

        return self

    @retry(HttpError, tries=4, delay=5, backoff=5, fatal_exceptions=(FatalHttpError,))
    def _execute(self, resource: Any):
        return resource.execute()

    def get_values_from_range(self, cell_range: str) -> ValueResponse:
        """Method to get data from a Spreadsheet range. Returns a value object which can be chained
        to a dataframe or list.

        Args:
            cell_range (str): The range of cells for which to get data.

        Returns:
            self (Spreadsheet): The object itself, to method chain.
        """

        get_response = self._execute(
            self.service.spreadsheets()
            .values()
            .batchGet(spreadsheetId=self.id, ranges=[cell_range])
        )

        value_range = get_response.get("valueRanges")[0]  # TODO: cleanup
        values = value_range.get("values", [])
        return ValueResponse(values)

    def to_list(self, cell_range: str) -> list:
        """Method to use to return a list of values.

        Returns:
            self._values (list): A list of the cell data gotten from get_values_from_range()
        """
        return self.get_values_from_range(cell_range=cell_range)

    def to_df(self, cell_range: str, header: bool = True) -> pd.DataFrame:
        """Method to use to return a pandas.DataFrame of values.

        Returns:
            df (pd.DataFrame): A pandas.DataFrame of the cell data gotten from get_values_from_range()
        """
        return self.get_values_from_range(cell_range=cell_range).to_df(header=header)

    def insert_data(
        self, insert_range: str, values: list, major_dimension: Dimension = Dimension.ROWS
    ):
        """Method to insert data from a list into a given range in the Spreadsheet.

        Args:
            insert_range (str): The cell range in which to insert data.
            values (list): The data values, as a list, to insert.
            major_dimension (Dimension): Indicates which dimension an operation should apply to.

        Returns:
            self (Spreadsheet): The object itself, to method chain.
        """

        value_range = {
            "range": insert_range,
            "majorDimension": major_dimension.value,
            "values": values,
        }

        self._values_update_queue.append(value_range)

        return self

    def insert_data_from_df(
        self, df: pd.DataFrame, insert_range: str, major_dimension: Dimension = Dimension.ROWS
    ):
        """Method to insert data from a pd.DataFrame into a given range in the Spreadsheet.

        Args:
            df (pd.DataFrame): Dataframe of data to insert.
            insert_range (str): The cell range in which to insert data.
            major_dimension (Dimension): Indicates which dimension an operation should apply to.

        Returns:
            self (Spreadsheet): The object itself, to method chain.
        """

        # TODO: this insert_data_from_df method might be better as a method that we can
        # attach to insert_data--something like: mySpreadsheet.insert_data().from_df()

        header = df.columns.values.tolist()
        data = df.values.tolist()

        values = []
        if len(header) > 0:
            values.append(header)
        values.extend(data)

        self.insert_data(insert_range=insert_range, values=values, major_dimension=major_dimension)

        return self

__init__(id, client=None)

Method to initialize the class.

The init method accepts a client connection to the Google API, which it uses to retrieve the properties of the spreadsheet and create a _spreadsheet object to execute other requests of the API.

Parameters:

Name Type Description Default
id str

id of the Google Spreadsheet. Spreadsheet id can be found in the Spreadsheet URL between /d/ and /edit, eg: https://docs.google.com/spreadsheets/d//edit#gid=0

 required
client googleapiclient.discovery.Resource

connection to the Google API Sheets resource.

 None
Source code in pygsuite/sheets/sheet.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def __init__(self, id: str, client: Optional[Resource] = None):
    """Method to initialize the class.

    The __init__ method accepts a client connection to the Google API, which it uses to retrieve the properties
    of the spreadsheet and create a _spreadsheet object to execute other requests of the API.

    Args:
        id (str): id of the Google Spreadsheet. Spreadsheet id can be found in the Spreadsheet URL between /d/ and /edit, eg:
            https://docs.google.com/spreadsheets/d/<spreadsheetId>/edit#gid=0
        client (googleapiclient.discovery.Resource): connection to the Google API Sheets resource.
    """

    self.service = client or Clients.sheets_client
    self.id = parse_id(id) if id else None
    DriveObject.__init__(self, id=id, client=client)

    self._spreadsheet = self._execute(self.service.spreadsheets().get(spreadsheetId=self.id))
    self._properties = self._spreadsheet.get("properties")

    # queues to add to and run in flush()
    self._spreadsheets_update_queue: List[Dict] = []
    self._values_update_queue: List[Dict] = []

create_sheet(sheet_properties=None, **kwargs)

Add a new sheet to the spreadsheet.

Source code in pygsuite/sheets/sheet.py 
205
206
207
208
209
210
def create_sheet(self, sheet_properties: Optional[SheetProperties] = None, **kwargs):
    """Add a new sheet to the spreadsheet."""
    sheet_properties = sheet_properties or SheetProperties(**kwargs)
    base = {"addSheet": {"properties": sheet_properties.to_json()}}
    self._spreadsheets_update_queue.append(base)
    return self

delete_sheet(title=None, id=None)

Method to delete a sheet by the title or id.

Parameters:

Name Type Description Default
title Optional

str): The title of a sheet to delete.

 None
id Optional

int): The id of a sheet to delete.

 None

Returns:

Name Type Description
self Spreadsheet

The object itself, to method chain.
Source code in pygsuite/sheets/sheet.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
def delete_sheet(self, title: Optional[str] = None, id: Optional[int] = None):
    """Method to delete a sheet by the title or id.

    Args:
        title (Optional: str): The title of a sheet to delete.
        id (Optional: int): The id of a sheet to delete.

    Returns:
        self (Spreadsheet): The object itself, to method chain.
    """

    # if an id is not given, title must be given; get id from title
    if not id:
        assert title is not None
        id = self[title].id

    base = {"deleteSheet": {"sheetId": id}}
    self._spreadsheets_update_queue.append(base)

    return self

flush(reverse=False, value_input_option=ValueInputOption.USER_ENTERED, include_values_in_response=False, response_value_render_option=ValueRenderOption.FORMATTED_VALUE, response_date_time_render_option=DateTimeRenderOption.SERIAL_NUMBER)

Method to execute the change queues created through other methods.

Parameters:

Name Type Description Default
reverse bool

If True, the changes queued will be executed in reverse order.

 False
value_input_option ValueInputOption

Indicates which dimension an operation should apply to.

 ValueInputOption.USER_ENTERED
include_values_in_response bool

If True, the API response will include the values changed.

 False
response_value_render_option ValueRenderOption

Determines how values should be rendered in the output.

 ValueRenderOption.FORMATTED_VALUE
response_date_time_render_option DateTimeRenderOption

Determines how dates should be rendered in the output.

 DateTimeRenderOption.SERIAL_NUMBER

Returns:

Name Type Description
response_dict dict

Dictionary of API response JSON.
Source code in pygsuite/sheets/sheet.py 
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
@retry((HttpError), tries=3, delay=10, backoff=5)
def flush(
    self,
    reverse: bool = False,
    value_input_option: ValueInputOption = ValueInputOption.USER_ENTERED,
    include_values_in_response: bool = False,
    response_value_render_option: ValueRenderOption = ValueRenderOption.FORMATTED_VALUE,
    response_date_time_render_option: DateTimeRenderOption = DateTimeRenderOption.SERIAL_NUMBER,
):
    """Method to execute the change queues created through other methods.

    Args:
        reverse (bool): If True, the changes queued will be executed in reverse order.
        value_input_option (ValueInputOption): Indicates which dimension an operation should apply to.
        include_values_in_response (bool): If True, the API response will include the values changed.
        response_value_render_option (ValueRenderOption): Determines how values should be rendered in the output.
        response_date_time_render_option (DateTimeRenderOption): Determines how dates should be rendered in the output.

    Returns:
        response_dict (dict): Dictionary of API response JSON.
    """

    if reverse:
        _spreadsheets_update_queue = list(reversed(self._spreadsheets_update_queue))
        _values_update_queue = list(reversed(self._values_update_queue))
    else:
        _spreadsheets_update_queue = self._spreadsheets_update_queue
        _values_update_queue = self._values_update_queue

    response_dict = dict()

    if len(_spreadsheets_update_queue) > 0:
        response_dict["spreadsheets_update_response"] = self._execute(
            self.service.spreadsheets().batchUpdate(
                body={"requests": _spreadsheets_update_queue}, spreadsheetId=self.id
            )
        ).get("responses")

    if len(_values_update_queue) > 0:
        response_dict["values_update_response"] = self._execute(
            self.service.spreadsheets()
            .values()
            .batchUpdate(
                spreadsheetId=self.id,
                body={
                    "valueInputOption": value_input_option.value,
                    "data": _values_update_queue,
                    "includeValuesInResponse": include_values_in_response,
                    "responseValueRenderOption": response_value_render_option.value,
                    "responseDateTimeRenderOption": response_date_time_render_option.value,
                },
            )
        ).get("responses")

    # the queues are reset in the refresh() method
    # self._spreadsheets_update_queue = []
    # self._values_update_queue = []
    self.refresh()

    return response_dict

get_values_from_range(cell_range)

Method to get data from a Spreadsheet range. Returns a value object which can be chained to a dataframe or list.

Parameters:

Name Type Description Default
cell_range str

The range of cells for which to get data.

 required

Returns:

Name Type Description
self Spreadsheet

The object itself, to method chain.
Source code in pygsuite/sheets/sheet.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
def get_values_from_range(self, cell_range: str) -> ValueResponse:
    """Method to get data from a Spreadsheet range. Returns a value object which can be chained
    to a dataframe or list.

    Args:
        cell_range (str): The range of cells for which to get data.

    Returns:
        self (Spreadsheet): The object itself, to method chain.
    """

    get_response = self._execute(
        self.service.spreadsheets()
        .values()
        .batchGet(spreadsheetId=self.id, ranges=[cell_range])
    )

    value_range = get_response.get("valueRanges")[0]  # TODO: cleanup
    values = value_range.get("values", [])
    return ValueResponse(values)

insert_data(insert_range, values, major_dimension=Dimension.ROWS)

Method to insert data from a list into a given range in the Spreadsheet.

Parameters:

Name Type Description Default
insert_range str

The cell range in which to insert data.

 required
values list

The data values, as a list, to insert.

 required
major_dimension Dimension

Indicates which dimension an operation should apply to.

 Dimension.ROWS

Returns:

Name Type Description
self Spreadsheet

The object itself, to method chain.
Source code in pygsuite/sheets/sheet.py 
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
def insert_data(
    self, insert_range: str, values: list, major_dimension: Dimension = Dimension.ROWS
):
    """Method to insert data from a list into a given range in the Spreadsheet.

    Args:
        insert_range (str): The cell range in which to insert data.
        values (list): The data values, as a list, to insert.
        major_dimension (Dimension): Indicates which dimension an operation should apply to.

    Returns:
        self (Spreadsheet): The object itself, to method chain.
    """

    value_range = {
        "range": insert_range,
        "majorDimension": major_dimension.value,
        "values": values,
    }

    self._values_update_queue.append(value_range)

    return self

insert_data_from_df(df, insert_range, major_dimension=Dimension.ROWS)

Method to insert data from a pd.DataFrame into a given range in the Spreadsheet.

Parameters:

Name Type Description Default
df pd.DataFrame

Dataframe of data to insert.

 required
insert_range str

The cell range in which to insert data.

 required
major_dimension Dimension

Indicates which dimension an operation should apply to.

 Dimension.ROWS

Returns:

Name Type Description
self Spreadsheet

The object itself, to method chain.
Source code in pygsuite/sheets/sheet.py 
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
def insert_data_from_df(
    self, df: pd.DataFrame, insert_range: str, major_dimension: Dimension = Dimension.ROWS
):
    """Method to insert data from a pd.DataFrame into a given range in the Spreadsheet.

    Args:
        df (pd.DataFrame): Dataframe of data to insert.
        insert_range (str): The cell range in which to insert data.
        major_dimension (Dimension): Indicates which dimension an operation should apply to.

    Returns:
        self (Spreadsheet): The object itself, to method chain.
    """

    # TODO: this insert_data_from_df method might be better as a method that we can
    # attach to insert_data--something like: mySpreadsheet.insert_data().from_df()

    header = df.columns.values.tolist()
    data = df.values.tolist()

    values = []
    if len(header) > 0:
        values.append(header)
    values.extend(data)

    self.insert_data(insert_range=insert_range, values=values, major_dimension=major_dimension)

    return self

refresh()

Method to refresh the spreadsheets API connection.

Source code in pygsuite/sheets/sheet.py 
137
138
139
140
141
142
def refresh(self):
    """Method to refresh the spreadsheets API connection."""
    self._spreadsheet = self._execute(self.service.spreadsheets().get(spreadsheetId=self.id))

    self._spreadsheets_update_queue = []
    self._values_update_queue = []

to_df(cell_range, header=True)

Method to use to return a pandas.DataFrame of values.

Returns:

Name Type Description
df pd.DataFrame

A pandas.DataFrame of the cell data gotten from get_values_from_range()
Source code in pygsuite/sheets/sheet.py 
288
289
290
291
292
293
294
def to_df(self, cell_range: str, header: bool = True) -> pd.DataFrame:
    """Method to use to return a pandas.DataFrame of values.

    Returns:
        df (pd.DataFrame): A pandas.DataFrame of the cell data gotten from get_values_from_range()
    """
    return self.get_values_from_range(cell_range=cell_range).to_df(header=header)

to_list(cell_range)

Method to use to return a list of values.

Returns:

Type Description
list

self._values (list): A list of the cell data gotten from get_values_from_range()
Source code in pygsuite/sheets/sheet.py 
280
281
282
283
284
285
286
def to_list(self, cell_range: str) -> list:
    """Method to use to return a list of values.

    Returns:
        self._values (list): A list of the cell data gotten from get_values_from_range()
    """
    return self.get_values_from_range(cell_range=cell_range)

worksheets() property

List of Worksheet objects for each worksheet in the Spreadsheet.

Source code in pygsuite/sheets/sheet.py 
132
133
134
135
@property
def worksheets(self):
    """List of Worksheet objects for each worksheet in the Spreadsheet."""
    return [Worksheet(sheet, self) for sheet in self._spreadsheet.get("sheets")]