Sheet

DateTimeRenderOption

Bases: Enum

DateTimeRenderOption: Determines how dates should be rendered in the output.

Google docs: https://developers.google.com/sheets/api/reference/rest/v4/DateTimeRenderOption

Source code in pygsuite/sheets/sheet.py

class DateTimeRenderOption(Enum):
    """DateTimeRenderOption: Determines how dates should be rendered in the output.

    Google docs: https://developers.google.com/sheets/api/reference/rest/v4/DateTimeRenderOption
    """

    SERIAL_NUMBER = "SERIAL_NUMBER"
    FORMATTED_STRING = "FORMATTED_STRING"

Dimension

Bases: Enum

Dimension: Indicates which dimension an operation should apply to.

Google docs: https://developers.google.com/sheets/api/reference/rest/v4/Dimension

Source code in pygsuite/sheets/sheet.py

class Dimension(Enum):
    """Dimension: Indicates which dimension an operation should apply to.

    Google docs: https://developers.google.com/sheets/api/reference/rest/v4/Dimension
    """

    ROWS = "ROWS"
    COLUMNS = "COLUMNS"

Spreadsheet

Bases: DriveObject

Base class for the GSuite Spreadsheets API.

Source code in pygsuite/sheets/sheet.py

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

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

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

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

@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

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

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

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

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

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

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

@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")]

ValueInputOption

Bases: Enum

ValueInputOption: Determines how data should be interpreted.

Google docs: https://developers.google.com/sheets/api/reference/rest/v4/ValueInputOption

Source code in pygsuite/sheets/sheet.py

class ValueInputOption(Enum):
    """ValueInputOption: Determines how data should be interpreted.

    Google docs: https://developers.google.com/sheets/api/reference/rest/v4/ValueInputOption
    """

    RAW = "RAW"
    USER_ENTERED = "USER_ENTERED"

ValueRenderOption

Bases: Enum

ValueRenderOption: Determines how values should be rendered in the output.

Google docs: https://developers.google.com/sheets/api/reference/rest/v4/ValueRenderOption

Source code in pygsuite/sheets/sheet.py

class ValueRenderOption(Enum):
    """ValueRenderOption: Determines how values should be rendered in the output.

    Google docs: https://developers.google.com/sheets/api/reference/rest/v4/ValueRenderOption
    """

    FORMATTED_VALUE = "FORMATTED_VALUE"
    UNFORMATTED_VALUE = "UNFORMATTED_VALUE"
    FORMULA = "FORMULA"

ValueResponse

Bases: list

Source code in pygsuite/sheets/sheet.py

class ValueResponse(list):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def to_list(self):
        return list(self)

    def to_df(self, header: bool):
        """Method to use with self.get_values_from_range(self, cell_range) to return a pandas.DataFrame of values.

        Returns:
            df (pd.DataFrame): A pandas.DataFrame of the cell data gotten from get_values_from_range()
        """
        if header:
            header = self[0]
            records = self[1:]
            df = pd.DataFrame.from_records(data=records, columns=header)
        else:
            records = self
            df = pd.DataFrame.from_records(data=records)

        return df

to_df(header)

Method to use with self.get_values_from_range(self, cell_range) 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

def to_df(self, header: bool):
    """Method to use with self.get_values_from_range(self, cell_range) to return a pandas.DataFrame of values.

    Returns:
        df (pd.DataFrame): A pandas.DataFrame of the cell data gotten from get_values_from_range()
    """
    if header:
        header = self[0]
        records = self[1:]
        df = pd.DataFrame.from_records(data=records, columns=header)
    else:
        records = self
        df = pd.DataFrame.from_records(data=records)

    return df