Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

bugfix/client export #1392

Merged
merged 4 commits into from
Jan 31, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
75 changes: 11 additions & 64 deletions gspread/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
from google.auth.credentials import Credentials
from requests import Response, Session

from .exceptions import APIError, SpreadsheetNotFound, UnSupportedExportFormat
from .exceptions import APIError, SpreadsheetNotFound
from .http_client import HTTPClient, HTTPClientType, ParamsType
from .spreadsheet import Spreadsheet
from .urls import (
Expand Down Expand Up @@ -239,15 +239,7 @@ def export(self, file_id: str, format: str = ExportFormat.PDF) -> bytes:
.. _ExportFormat: https://developers.google.com/drive/api/guides/ref-export-formats
"""

if format not in ExportFormat:
raise UnSupportedExportFormat

url = "{}/{}/export".format(DRIVE_FILES_API_V3_URL, file_id)

params: ParamsType = {"mimeType": format}

r = self.http_client.request("get", url, params=params)
return r.content
return self.http_client.export(file_id=file_id, format=format)

def copy(
self,
Expand Down Expand Up @@ -406,34 +398,14 @@ def list_permissions(self, file_id: str) -> List[Dict[str, Union[str, bool]]]:

:param str file_id: a spreadsheet ID (aka file ID).
"""
url = "{}/{}/permissions".format(DRIVE_FILES_API_V3_URL, file_id)

params: ParamsType = {
"supportsAllDrives": True,
"fields": "nextPageToken,permissions",
}

token = ""

permissions = []

while token is not None:
if token:
params["pageToken"] = token

r = self.http_client.request("get", url, params=params).json()
permissions.extend(r["permissions"])

token = r.get("nextPageToken", None)

return permissions
return self.http_client.list_permissions(file_id)

def insert_permission(
self,
file_id: str,
value: Optional[str],
perm_type: Optional[str],
role: Optional[str],
value: Optional[str] = None,
perm_type: Optional[str] = None,
role: Optional[str] = None,
notify: bool = True,
email_message: Optional[str] = None,
with_link: bool = False,
Expand All @@ -442,7 +414,7 @@ def insert_permission(

:param str file_id: a spreadsheet ID (aka file ID).
:param value: user or group e-mail address, domain name
or None for 'default' type.
or None for 'anyone' type.
:type value: str, None
:param str perm_type: (optional) The account type.
Allowed values are: ``user``, ``group``, ``domain``, ``anyone``
Expand Down Expand Up @@ -478,39 +450,14 @@ def insert_permission(
)

"""

url = "{}/{}/permissions".format(DRIVE_FILES_API_V3_URL, file_id)
payload = {
"type": perm_type,
"role": role,
"withLink": with_link,
}
params: ParamsType = {
"supportsAllDrives": "true",
}

if perm_type == "domain":
payload["domain"] = value
elif perm_type in {"user", "group"}:
payload["emailAddress"] = value
params["sendNotificationEmail"] = notify
params["emailMessage"] = email_message
elif perm_type == "anyone":
pass
else:
raise ValueError("Invalid permission type: {}".format(perm_type))

return self.http_client.request("post", url, json=payload, params=params)
return self.http_client.insert_permission(
file_id, value, perm_type, role, notify, email_message, with_link
)

def remove_permission(self, file_id: str, permission_id: str) -> None:
"""Deletes a permission from a file.

:param str file_id: a spreadsheet ID (aka file ID.)
:param str permission_id: an ID for the permission.
"""
url = "{}/{}/permissions/{}".format(
DRIVE_FILES_API_V3_URL, file_id, permission_id
)

params: ParamsType = {"supportsAllDrives": True}
self.http_client.request("delete", url, params=params)
self.http_client.remove_permission(file_id, permission_id)
164 changes: 161 additions & 3 deletions gspread/http_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,13 +7,24 @@

"""
from http import HTTPStatus
from typing import IO, Any, List, Mapping, MutableMapping, Optional, Tuple, Type, Union
from typing import (
IO,
Any,
Dict,
List,
Mapping,
MutableMapping,
Optional,
Tuple,
Type,
Union,
)

from google.auth.credentials import Credentials
from google.auth.transport.requests import AuthorizedSession
from requests import Response, Session

from .exceptions import APIError
from .exceptions import APIError, UnSupportedExportFormat
from .urls import (
DRIVE_FILES_API_V3_URL,
SPREADSHEET_BATCH_UPDATE_URL,
Expand All @@ -26,7 +37,7 @@
SPREADSHEET_VALUES_CLEAR_URL,
SPREADSHEET_VALUES_URL,
)
from .utils import convert_credentials, quote
from .utils import ExportFormat, convert_credentials, quote

ParamsType = MutableMapping[str, Optional[Union[str, int, bool, float, List[str]]]]

Expand Down Expand Up @@ -306,6 +317,153 @@ def get_file_drive_metadata(self, id: str) -> Any:

return res.json()

def export(self, file_id: str, format: str = ExportFormat.PDF) -> bytes:
"""Export the spreadsheet in the given format.

:param str file_id: The key of the spreadsheet to export

:param str format: The format of the resulting file.
Possible values are:

* ``ExportFormat.PDF``
* ``ExportFormat.EXCEL``
* ``ExportFormat.CSV``
* ``ExportFormat.OPEN_OFFICE_SHEET``
* ``ExportFormat.TSV``
* ``ExportFormat.ZIPPED_HTML``

See `ExportFormat`_ in the Drive API.

:type format: :class:`~gspread.utils.ExportFormat`

:returns bytes: The content of the exported file.

.. _ExportFormat: https://developers.google.com/drive/api/guides/ref-export-formats
"""

if format not in ExportFormat:
raise UnSupportedExportFormat

url = "{}/{}/export".format(DRIVE_FILES_API_V3_URL, file_id)

params: ParamsType = {"mimeType": format}

r = self.request("get", url, params=params)
return r.content

def insert_permission(
self,
file_id: str,
email_address: Optional[str],
perm_type: Optional[str],
role: Optional[str],
notify: bool = True,
email_message: Optional[str] = None,
with_link: bool = False,
) -> Response:
"""Creates a new permission for a file.

:param str file_id: a spreadsheet ID (aka file ID).
:param email_address: user or group e-mail address, domain name
or None for 'anyone' type.
:type email_address: str, None
:param str perm_type: (optional) The account type.
Allowed values are: ``user``, ``group``, ``domain``, ``anyone``
:param str role: (optional) The primary role for this user.
Allowed values are: ``owner``, ``writer``, ``reader``
:param bool notify: Whether to send an email to the target
user/domain. Default ``True``.
:param str email_message: (optional) An email message to be sent
if ``notify=True``.
:param bool with_link: Whether the link is required for this
permission to be active. Default ``False``.

:returns dict: the newly created permission

Examples::

# Give write permissions to otto@example.com

gc.insert_permission(
'0BmgG6nO_6dprnRRUWl1UFE',
'otto@example.org',
perm_type='user',
role='writer'
)

# Make the spreadsheet publicly readable

gc.insert_permission(
'0BmgG6nO_6dprnRRUWl1UFE',
None,
perm_type='anyone',
role='reader'
)

"""
url = "{}/{}/permissions".format(DRIVE_FILES_API_V3_URL, file_id)
payload = {
"type": perm_type,
"role": role,
"withLink": with_link,
}
params: ParamsType = {
"supportsAllDrives": "true",
}

if perm_type == "domain":
payload["domain"] = email_address
elif perm_type in {"user", "group"}:
payload["emailAddress"] = email_address
params["sendNotificationEmail"] = notify
params["emailMessage"] = email_message
elif perm_type == "anyone":
pass
else:
raise ValueError("Invalid permission type: {}".format(perm_type))

return self.request("post", url, json=payload, params=params)

def list_permissions(self, file_id: str) -> List[Dict[str, Union[str, bool]]]:
"""Retrieve a list of permissions for a file.

:param str file_id: a spreadsheet ID (aka file ID).
"""
url = "{}/{}/permissions".format(DRIVE_FILES_API_V3_URL, file_id)

params: ParamsType = {
"supportsAllDrives": True,
"fields": "nextPageToken,permissions",
}

token = ""

permissions = []

while token is not None:
if token:
params["pageToken"] = token

r = self.request("get", url, params=params).json()
permissions.extend(r["permissions"])

token = r.get("nextPageToken", None)

return permissions

def remove_permission(self, file_id: str, permission_id: str) -> None:
"""Deletes a permission from a file.

:param str file_id: a spreadsheet ID (aka file ID.)
:param str permission_id: an ID for the permission.
"""
url = "{}/{}/permissions/{}".format(
DRIVE_FILES_API_V3_URL, file_id, permission_id
)

params: ParamsType = {"supportsAllDrives": True}
self.request("delete", url, params=params)


class BackOffHTTPClient(HTTPClient):
"""BackoffClient is a gspread client with exponential
Expand Down
8 changes: 4 additions & 4 deletions gspread/spreadsheet.py
Original file line number Diff line number Diff line change
Expand Up @@ -457,9 +457,9 @@ def share(
):
"""Share the spreadsheet with other accounts.

:param value: user or group e-mail address, domain name
or None for 'default' type.
:type value: str, None
:param email_address: user or group e-mail address, domain name
or None for 'anyone' type.
:type email_address: str, None
:param perm_type: The account type.
Allowed values are: ``user``, ``group``, ``domain``,
``anyone``.
Expand All @@ -484,7 +484,7 @@ def share(
"""
return self.client.insert_permission(
self.id,
value=email_address,
email_address=email_address,
perm_type=perm_type,
role=role,
notify=notify,
Expand Down
Loading