PyGenericSpreadSheet API Documentation

googlesheet

Created on 2022-04-18

@author: wf

GoogleSheet

Bases: object

GoogleSheet Handling

Source code in spreadsheet/googlesheet.py

class GoogleSheet(object):
    """
    GoogleSheet Handling
    """

    def __init__(self, url: str, readonly: bool = True, max_retries:int=5, max_wait:float=60.0):
        """
        Initializes an instance of GoogleSheet.

        Args:
            url (str): URL to the Google Sheet.
            readonly (bool): If True, uses read-only scopes, otherwise uses full access scopes.
        """
        self.url = url
        self.sheet_dict = {}
        self.scopes = ['https://www.googleapis.com/auth/spreadsheets.readonly'] \
                      if readonly else \
                      ['https://www.googleapis.com/auth/spreadsheets']
        self.sheet_dict = {}
        self.max_retries=max_retries
        self.max_wait=max_wait
        self.credentials = self.get_credentials()

    def safe_api_call(self, func, *args, **kwargs):
        """
        Safe wrapper for API calls with exponential backoff.

        Args:
            func: The API function to call.
            *args: Arguments for the API function.
            **kwargs: Keyword arguments for the API function.

        Returns:
            The result of the API function call.
        """
        base_sleep = 1  # Initial sleep time in seconds
        total_sleep = 0  # Total time slept

        for attempt in range(self.max_retries):
            try:
                return func(*args, **kwargs)
            except gspread.exceptions.APIError as e:
                if e.response.status_code != 429:
                    raise  # Reraise if not a quota limit error

                # Calculate sleep time with exponential backoff
                sleep_time = min(base_sleep * (2 ** attempt) + random.random(), self.max_wait - total_sleep)
                if total_sleep + sleep_time > self.max_wait:
                    raise Exception(f"Exceeded maximum wait time of {self.max_wait} seconds for Google API calls.") from e

                print(f"Quota exceeded, retrying in {sleep_time:.2f} seconds.")
                time.sleep(sleep_time)
                total_sleep += sleep_time
        raise Exception("Maximum retries reached without success.")

    def get_credentials(self):
        """
        Check for Google API credentials in the home directory or
        the GOOGLE_API_KEY environment variable
        """
        google_api_key_json=os.getenv("GOOGLE_API_KEY")
        credentials=None
        if google_api_key_json:
            creds_dict=json.loads(google_api_key_json)
            credentials=Credentials.from_service_account_info(creds_dict, scopes=self.scopes)
        else:
            cred_path = os.path.join(os.path.expanduser("~"), ".ose", "google-api-key.json")
            if os.path.exists(cred_path):
                credentials = Credentials.from_service_account_file(cred_path, scopes=self.scopes)
        return credentials


    def open(self, sheet_names: list = None) -> dict:
        """
        Opens the Google Sheet and loads the data from specified sheet names into a dictionary.

        Args:
            sheet_names: Optional list of sheet names to open. Opens all sheets if None.

        Returns:
            A dictionary with sheet names as keys and lists of dictionaries (rows) as values.
        """
        credentials = self.get_credentials()
        if not credentials:
            raise Exception("Credentials not found.")

        self.gc = gspread.authorize(credentials)
        self.sh = self.safe_api_call(self.gc.open_by_url, self.url)
        # Retrieve all sheet names if none are provided
        sheet_names = sheet_names or [sheet.title for sheet in self.sh.worksheets()]

        for sheet_name in sheet_names:
            worksheet = self.sh.worksheet(sheet_name)
            records = self.safe_api_call(worksheet.get_all_records)
            self.sheet_dict[sheet_name] = records

        return self.sheet_dict

    def asListOfDicts(self, sheet_name: str)->List:
        """
        Converts a sheet to a list of dictionaries.

        Args:
            sheet_name (str): The name of the sheet to convert.

        Returns:
            A list of dictionaries, each representing a row in the sheet.
        """
        if not sheet_name in self.sheet_dict:
            self.open[sheet_name]
        lod = self.sheet_dict.get(sheet_name)
        return lod

    def fixRows(self, lod: list):
        """
        fix Rows by filtering unnamed columns and trimming
        column names
        """
        for row in lod:
            for key in list(row.keys()):
                if key.startswith("Unnamed"):
                    del row[key]
                trimmedKey = key.strip()
                if trimmedKey != key:
                    value = row[key]
                    row[trimmedKey] = value
                    del row[key]

    @classmethod
    def toWikibaseQuery(
        cls, url: str, sheetName: str = "WikidataMapping", debug: bool = False
    ) -> Dict[str, "WikibaseQuery"]:
        """
        create a dict of wikibaseQueries from the given google sheets row descriptions

        Args:
            url(str): the url of the sheet
            sheetName(str): the name of the sheet with the description
            debug(bool): if True switch on debugging
        """
        gs = GoogleSheet(url)
        gs.open([sheetName])
        entityMapRows = gs.asListOfDicts(sheetName)
        queries=WikibaseQuery.ofMapRows(entityMapRows, debug=debug)
        return queries

    @classmethod
    def toSparql(
        cls,
        url: str,
        sheetName: str,
        entityName: str,
        pkColumn: str,
        mappingSheetName: str = "WikidataMapping",
        lang: str = "en",
        debug: bool = False,
    ) -> ("WikibaseQuery", str):
        """
        get a sparql query for the given google sheet

        Args:
            url (str): the url of the sheet
            sheetName (str): the name of the sheet with the description
            entityName (str): the name of the entity as defined in the Wikidata mapping
            pkColumn (str): the column to use as a "primary key"
            mappingSheetName (str): the name of the sheet with the Wikidata mappings
            lang (str): the language to use (if any)
            debug (bool): if True switch on debugging

        Returns:
            WikibaseQuery
        """
        queries = cls.toWikibaseQuery(url, mappingSheetName, debug)
        gs = GoogleSheet(url)
        gs.open([sheetName])
        lod = gs.asListOfDicts(sheetName)
        lodByPk, _dup = LOD.getLookup(lod, pkColumn)
        query = queries[entityName]
        propRow = query.propertiesByColumn[pkColumn]
        pk = propRow["PropertyName"]
        pkVarname = propRow["PropVarname"]
        pkType = propRow["Type"]
        valuesClause = query.getValuesClause(
            lodByPk.keys(), propVarname=pkVarname, propType=pkType, lang=lang
        )

        sparql = query.asSparql(
            filterClause=valuesClause, orderClause=f"ORDER BY ?{pkVarname}", pk=pk
        )
        return query, sparql

__init__(url, readonly=True, max_retries=5, max_wait=60.0)

Initializes an instance of GoogleSheet.

Parameters:

Name	Type	Description	Default
url	str	URL to the Google Sheet.	required
readonly	bool	If True, uses read-only scopes, otherwise uses full access scopes.	True

Source code in spreadsheet/googlesheet.py

def __init__(self, url: str, readonly: bool = True, max_retries:int=5, max_wait:float=60.0):
    """
    Initializes an instance of GoogleSheet.

    Args:
        url (str): URL to the Google Sheet.
        readonly (bool): If True, uses read-only scopes, otherwise uses full access scopes.
    """
    self.url = url
    self.sheet_dict = {}
    self.scopes = ['https://www.googleapis.com/auth/spreadsheets.readonly'] \
                  if readonly else \
                  ['https://www.googleapis.com/auth/spreadsheets']
    self.sheet_dict = {}
    self.max_retries=max_retries
    self.max_wait=max_wait
    self.credentials = self.get_credentials()

asListOfDicts(sheet_name)

Converts a sheet to a list of dictionaries.

Parameters:

Name	Type	Description	Default
sheet_name	str	The name of the sheet to convert.	required

Returns:

Type	Description
List	A list of dictionaries, each representing a row in the sheet.

Source code in spreadsheet/googlesheet.py

def asListOfDicts(self, sheet_name: str)->List:
    """
    Converts a sheet to a list of dictionaries.

    Args:
        sheet_name (str): The name of the sheet to convert.

    Returns:
        A list of dictionaries, each representing a row in the sheet.
    """
    if not sheet_name in self.sheet_dict:
        self.open[sheet_name]
    lod = self.sheet_dict.get(sheet_name)
    return lod

fixRows(lod)

fix Rows by filtering unnamed columns and trimming column names

Source code in spreadsheet/googlesheet.py

def fixRows(self, lod: list):
    """
    fix Rows by filtering unnamed columns and trimming
    column names
    """
    for row in lod:
        for key in list(row.keys()):
            if key.startswith("Unnamed"):
                del row[key]
            trimmedKey = key.strip()
            if trimmedKey != key:
                value = row[key]
                row[trimmedKey] = value
                del row[key]

get_credentials()

Check for Google API credentials in the home directory or the GOOGLE_API_KEY environment variable

Source code in spreadsheet/googlesheet.py

def get_credentials(self):
    """
    Check for Google API credentials in the home directory or
    the GOOGLE_API_KEY environment variable
    """
    google_api_key_json=os.getenv("GOOGLE_API_KEY")
    credentials=None
    if google_api_key_json:
        creds_dict=json.loads(google_api_key_json)
        credentials=Credentials.from_service_account_info(creds_dict, scopes=self.scopes)
    else:
        cred_path = os.path.join(os.path.expanduser("~"), ".ose", "google-api-key.json")
        if os.path.exists(cred_path):
            credentials = Credentials.from_service_account_file(cred_path, scopes=self.scopes)
    return credentials

open(sheet_names=None)

Opens the Google Sheet and loads the data from specified sheet names into a dictionary.

Parameters:

Name	Type	Description	Default
sheet_names	list	Optional list of sheet names to open. Opens all sheets if None.	None

Returns:

Type	Description
dict	A dictionary with sheet names as keys and lists of dictionaries (rows) as values.

Source code in spreadsheet/googlesheet.py

def open(self, sheet_names: list = None) -> dict:
    """
    Opens the Google Sheet and loads the data from specified sheet names into a dictionary.

    Args:
        sheet_names: Optional list of sheet names to open. Opens all sheets if None.

    Returns:
        A dictionary with sheet names as keys and lists of dictionaries (rows) as values.
    """
    credentials = self.get_credentials()
    if not credentials:
        raise Exception("Credentials not found.")

    self.gc = gspread.authorize(credentials)
    self.sh = self.safe_api_call(self.gc.open_by_url, self.url)
    # Retrieve all sheet names if none are provided
    sheet_names = sheet_names or [sheet.title for sheet in self.sh.worksheets()]

    for sheet_name in sheet_names:
        worksheet = self.sh.worksheet(sheet_name)
        records = self.safe_api_call(worksheet.get_all_records)
        self.sheet_dict[sheet_name] = records

    return self.sheet_dict

safe_api_call(func, *args, **kwargs)

Safe wrapper for API calls with exponential backoff.

Parameters:

Name	Type	Description	Default
func		The API function to call.	required
*args		Arguments for the API function.	()
**kwargs		Keyword arguments for the API function.	{}

Returns:

Type	Description
	The result of the API function call.

Source code in spreadsheet/googlesheet.py

def safe_api_call(self, func, *args, **kwargs):
    """
    Safe wrapper for API calls with exponential backoff.

    Args:
        func: The API function to call.
        *args: Arguments for the API function.
        **kwargs: Keyword arguments for the API function.

    Returns:
        The result of the API function call.
    """
    base_sleep = 1  # Initial sleep time in seconds
    total_sleep = 0  # Total time slept

    for attempt in range(self.max_retries):
        try:
            return func(*args, **kwargs)
        except gspread.exceptions.APIError as e:
            if e.response.status_code != 429:
                raise  # Reraise if not a quota limit error

            # Calculate sleep time with exponential backoff
            sleep_time = min(base_sleep * (2 ** attempt) + random.random(), self.max_wait - total_sleep)
            if total_sleep + sleep_time > self.max_wait:
                raise Exception(f"Exceeded maximum wait time of {self.max_wait} seconds for Google API calls.") from e

            print(f"Quota exceeded, retrying in {sleep_time:.2f} seconds.")
            time.sleep(sleep_time)
            total_sleep += sleep_time
    raise Exception("Maximum retries reached without success.")

toSparql(url, sheetName, entityName, pkColumn, mappingSheetName='WikidataMapping', lang='en', debug=False) classmethod

get a sparql query for the given google sheet

Parameters:

Name	Type	Description	Default
url	str	the url of the sheet	required
sheetName	str	the name of the sheet with the description	required
entityName	str	the name of the entity as defined in the Wikidata mapping	required
pkColumn	str	the column to use as a "primary key"	required
mappingSheetName	str	the name of the sheet with the Wikidata mappings	'WikidataMapping'
lang	str	the language to use (if any)	'en'
debug	bool	if True switch on debugging	False

Returns:

Type	Description
(WikibaseQuery, str)	WikibaseQuery

Source code in spreadsheet/googlesheet.py

@classmethod
def toSparql(
    cls,
    url: str,
    sheetName: str,
    entityName: str,
    pkColumn: str,
    mappingSheetName: str = "WikidataMapping",
    lang: str = "en",
    debug: bool = False,
) -> ("WikibaseQuery", str):
    """
    get a sparql query for the given google sheet

    Args:
        url (str): the url of the sheet
        sheetName (str): the name of the sheet with the description
        entityName (str): the name of the entity as defined in the Wikidata mapping
        pkColumn (str): the column to use as a "primary key"
        mappingSheetName (str): the name of the sheet with the Wikidata mappings
        lang (str): the language to use (if any)
        debug (bool): if True switch on debugging

    Returns:
        WikibaseQuery
    """
    queries = cls.toWikibaseQuery(url, mappingSheetName, debug)
    gs = GoogleSheet(url)
    gs.open([sheetName])
    lod = gs.asListOfDicts(sheetName)
    lodByPk, _dup = LOD.getLookup(lod, pkColumn)
    query = queries[entityName]
    propRow = query.propertiesByColumn[pkColumn]
    pk = propRow["PropertyName"]
    pkVarname = propRow["PropVarname"]
    pkType = propRow["Type"]
    valuesClause = query.getValuesClause(
        lodByPk.keys(), propVarname=pkVarname, propType=pkType, lang=lang
    )

    sparql = query.asSparql(
        filterClause=valuesClause, orderClause=f"ORDER BY ?{pkVarname}", pk=pk
    )
    return query, sparql

toWikibaseQuery(url, sheetName='WikidataMapping', debug=False) classmethod

create a dict of wikibaseQueries from the given google sheets row descriptions

Parameters:

Name	Type	Description	Default
url(str)		the url of the sheet	required
sheetName(str)		the name of the sheet with the description	required
debug(bool)		if True switch on debugging	required

Source code in spreadsheet/googlesheet.py

@classmethod
def toWikibaseQuery(
    cls, url: str, sheetName: str = "WikidataMapping", debug: bool = False
) -> Dict[str, "WikibaseQuery"]:
    """
    create a dict of wikibaseQueries from the given google sheets row descriptions

    Args:
        url(str): the url of the sheet
        sheetName(str): the name of the sheet with the description
        debug(bool): if True switch on debugging
    """
    gs = GoogleSheet(url)
    gs.open([sheetName])
    entityMapRows = gs.asListOfDicts(sheetName)
    queries=WikibaseQuery.ofMapRows(entityMapRows, debug=debug)
    return queries

spreadsheet

CSVSpreadSheet

Bases: SpreadSheet

CSV Spreadsheet packaging as ZIP file of CSV files

Source code in spreadsheet/spreadsheet.py

class CSVSpreadSheet(SpreadSheet):
    """
    CSV Spreadsheet packaging as ZIP file of CSV files
    """

    FILE_TYPE = ".zip"
    TABLE_TYPE = ".csv"
    MIME_TYPE = "application/zip"

    def __init__(self, name: str):
        super().__init__(name=name, spreadSheetType=SpreadSheetType.CSV)

    def toBytesIO(self) -> BytesIO:
        """
        Converts the document into an BytesIO stream

        Returns:
            BytesIO Stream of the document
        """
        buffer = BytesIO()
        buffer.name = self.filename
        with ZipFile(buffer, mode="w") as documentZip:
            for tableName, table in self.tables.items():
                csv = CSV.toCSV(table)
                documentZip.writestr(tableName + self.TABLE_TYPE, csv)
        buffer.seek(0)
        return buffer

    def _loadFromBuffer(self, file):
        """
        load the document from the given .zip file
        Args:
            file: absolut file path to the file that should be loaded
        Returns:

        """
        fileName = file.name
        tables = {}
        if fileName.endswith(self.FILE_TYPE):
            with ZipFile(file, mode="r") as documentZip:
                archivedFiles = documentZip.namelist()
                for archivedFile in archivedFiles:
                    with documentZip.open(archivedFile) as csvFile:
                        lod = CSV.fromCSV(csvFile.read().decode())
                        tableName = archivedFile[: -len(self.TABLE_TYPE)]
                        tables[tableName] = lod
        elif fileName.endswith(self.TABLE_TYPE):
            # single csv file load as sheet with one table
            lod = CSV.fromCSV(file.read().decode())
            tableName = fileName[: -len(self.TABLE_TYPE)]
            tables[tableName] = lod
        return tables

toBytesIO()

Converts the document into an BytesIO stream

Returns:

Type	Description
BytesIO	BytesIO Stream of the document

Source code in spreadsheet/spreadsheet.py

def toBytesIO(self) -> BytesIO:
    """
    Converts the document into an BytesIO stream

    Returns:
        BytesIO Stream of the document
    """
    buffer = BytesIO()
    buffer.name = self.filename
    with ZipFile(buffer, mode="w") as documentZip:
        for tableName, table in self.tables.items():
            csv = CSV.toCSV(table)
            documentZip.writestr(tableName + self.TABLE_TYPE, csv)
    buffer.seek(0)
    return buffer

ExcelDocument

Bases: SpreadSheet

Provides methods to convert LoDs to an excel document and vice versa

Source code in spreadsheet/spreadsheet.py

class ExcelDocument(SpreadSheet):
    """
    Provides methods to convert LoDs to an excel document and vice versa
    """

    FILE_TYPE = ".xlsx"
    MIME_TYPE = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"

    def __init__(
        self,
        name: str,
        engine: str = None,
        spreadSheetType: SpreadSheetType = SpreadSheetType.EXCEL,
    ):
        """
        Args:
            name(str): name of the document
        """
        super().__init__(name=name, spreadSheetType=spreadSheetType)
        if engine is None:
            engine = "xlsxwriter"
        self.engine_kwargs = {"options": {"strings_to_numbers": True}}

        self.engine = engine

    def toBytesIO(self) -> BytesIO:
        """
        Converts the document into an BytesIO stream

        Returns:
            BytesIO Stream of the document
        """
        buffer = BytesIO()

        with pd.ExcelWriter(
            buffer,
            engine=self.engine,
            engine_kwargs=self.engine_kwargs,
        ) as writer:
            for tableName, tableData in self.tables.items():
                df = pd.DataFrame(tableData)
                df.to_excel(writer, sheet_name=tableName, index=False)
        buffer.seek(0)
        buffer.name = self.filename
        return buffer

    def _loadFromBuffer(self, buffer):
        """
        read my table from the given BytesIO buffer
        """
        sheets = pd.read_excel(buffer, sheet_name=None).keys()
        tables = {}
        for sheet in sheets:
            df = pd.read_excel(buffer, sheet_name=sheet, na_values=None)
            df = df.where(pd.notnull(df), None)
            lod = df.to_dict("records")
            # NaT handling issue due to a bug in pandas https://github.com/pandas-dev/pandas/issues/29024
            lod = [
                {
                    k: v.to_pydatetime()
                    if isinstance(v, Timestamp)
                    else None
                    if isinstance(v, type(NaT))
                    else v
                    for k, v in d.items()
                }
                for d in lod
            ]
            # float nan to None
            lod = [
                {
                    k: v if not (isinstance(v, float) and math.isnan(v)) else None
                    for k, v in d.items()
                }
                for d in lod
            ]
            tables[sheet] = lod
        return tables

__init__(name, engine=None, spreadSheetType=SpreadSheetType.EXCEL)

Parameters:

Name	Type	Description	Default
name(str)		name of the document	required

Source code in spreadsheet/spreadsheet.py

def __init__(
    self,
    name: str,
    engine: str = None,
    spreadSheetType: SpreadSheetType = SpreadSheetType.EXCEL,
):
    """
    Args:
        name(str): name of the document
    """
    super().__init__(name=name, spreadSheetType=spreadSheetType)
    if engine is None:
        engine = "xlsxwriter"
    self.engine_kwargs = {"options": {"strings_to_numbers": True}}

    self.engine = engine

toBytesIO()

Converts the document into an BytesIO stream

Returns:

Type	Description
BytesIO	BytesIO Stream of the document

Source code in spreadsheet/spreadsheet.py

def toBytesIO(self) -> BytesIO:
    """
    Converts the document into an BytesIO stream

    Returns:
        BytesIO Stream of the document
    """
    buffer = BytesIO()

    with pd.ExcelWriter(
        buffer,
        engine=self.engine,
        engine_kwargs=self.engine_kwargs,
    ) as writer:
        for tableName, tableData in self.tables.items():
            df = pd.DataFrame(tableData)
            df.to_excel(writer, sheet_name=tableName, index=False)
    buffer.seek(0)
    buffer.name = self.filename
    return buffer

Format

potential Formats

Source code in spreadsheet/spreadsheet.py

class Format:
    """
    potential Formats
    """

    formatMap = {
        "CSV": {
            "name": "CSV",
            "title": "Comma separated Values",
            "postfix": ".csv",
            "mimetype": "text/csv",
        },
        "EXCEL": {
            "name": "Excel",
            "title": "Microsoft Excel",
            "postfix": ".xlsx",
            "mimetype": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        },
        "JSON": {
            "name": "JSON",
            "title": "Javascript Simple Object Notation",
            "postfix": ".json",
            "mimetype": "application/json",
        },
        "ODS": {
            "name": "ODS",
            "title": "OpenDocument Spreadsheet",
            "postfix": ".ods",
            "mimetype": "application/vnd.oasis.opendocument.onlinespreadsheet",
        },
    }

OdsDocument

Bases: ExcelDocument

OpenDocument Spreadsheet that can store multiple tables. Provides functions to traverse between LoD and ODS document

Source code in spreadsheet/spreadsheet.py

class OdsDocument(ExcelDocument):
    """
    OpenDocument Spreadsheet that can store multiple tables.
    Provides functions to traverse between LoD and ODS document
    """

    FILE_TYPE = ".ods"
    MIME_TYPE = "application/vnd.oasis.opendocument.onlinespreadsheet"

    def __init__(self, name: str):
        """
        Args:
            name(str): name of the document
        """

        super().__init__(name=name, spreadSheetType=SpreadSheetType.ODS, engine="odf")
        self.engine_kwargs = {}

__init__(name)

Parameters:

Name	Type	Description	Default
name(str)		name of the document	required

Source code in spreadsheet/spreadsheet.py

def __init__(self, name: str):
    """
    Args:
        name(str): name of the document
    """

    super().__init__(name=name, spreadSheetType=SpreadSheetType.ODS, engine="odf")
    self.engine_kwargs = {}

SpreadSheet

i am onlinespreadsheet

Source code in spreadsheet/spreadsheet.py

class SpreadSheet:
    """
    i am onlinespreadsheet
    """

    FILE_TYPE = NotImplemented
    MIME_TYPE = NotImplemented

    def __init__(self, name: str, spreadSheetType: SpreadSheetType):
        """
        constructor
        """
        self.name = name
        self.spreadSheetType = spreadSheetType
        self.tables = {}  # dict of lods
        pass

    @classmethod
    def create(cls, spreadSheetType: SpreadSheetType, name: str):
        """
        create a SpreadSheet of the given types

        Args:
            spreadSheetType(SpreadSheetType): the type of onlinespreadsheet to create
            name(str): the name of the onlinespreadsheet

        """
        spreadSheet = None
        if spreadSheetType == SpreadSheetType.EXCEL:
            spreadSheet = ExcelDocument(name=name)
        elif spreadSheetType == SpreadSheetType.ODS:
            spreadSheet = OdsDocument(name=name)
        elif spreadSheetType == SpreadSheetType.CSV:
            spreadSheet = CSVSpreadSheet(name=name)
        return spreadSheet

    @classmethod
    def load(cls, document):
        """
        Tries to load the given document as SpreadSheet
        Args:
            document: onlinespreadsheet document to load

        Returns:
            SpreadSheet
        """
        documentSpreadSheetType = None
        documentName = ""
        spreadsheet = None
        # TODO - use SpreadSheeeType enum instead
        spreadSheetTypes = [OdsDocument, ExcelDocument, CSVSpreadSheet]
        # TODO Get rid of Werkzeug dependency ...
        # if isinstance(document, FileStorage):
        #    document.stream.seek(0)
        #    for spreadSheetType in spreadSheetTypes:
        #        if document.filename.endswith(spreadSheetType.FILE_TYPE):
        #            documentName=document.filename[:-len(spreadSheetType.FILE_TYPE)]
        #            documentSpreadSheetType=spreadSheetType
        #            break
        if (
            isinstance(document, io.BytesIO)
            or isinstance(document, io.StringIO)
            or isinstance(document, io.TextIOWrapper)
        ):
            document.seek(0)
            for spreadSheetType in spreadSheetTypes:
                if document.name.endswith(spreadSheetType.FILE_TYPE):
                    documentName = document.name[: -len(spreadSheetType.FILE_TYPE)]
                    documentSpreadSheetType = spreadSheetType
                    break
        elif isinstance(document, str):
            try:
                buffer = None
                with open(document, "rb") as f:
                    buffer = io.BytesIO(f.read())
                    buffer.name = document
                return cls.load(buffer)
            except Exception as e:
                print(e)
                raise e
        else:
            print("Unable to load SpreadSheet")
            return None
        if documentSpreadSheetType:
            spreadsheet = documentSpreadSheetType(name=documentName)
            spreadsheet.loadFromFile(document)
        return spreadsheet

    def getTable(self, name: str):
        """
        returns the data corresponding to the given table name
        Args:
            name: name of the table

        Returns:
            LoD
        """
        if name in self.tables:
            return self.tables[name]

    def addTable(self, name: str, lod: list, headers: dict = None):
        """
        add the given data as table to the document

        Args:
            name(str): name of the table
            lod: data that should be added to the document as table
            headers(dict): Mapping from dict key to the new headers. Also functions as restriction. If not defined dict key are used as headers
        """
        if headers:
            lod = [
                {
                    newHeader: record.get(oldHeader, None)
                    for oldHeader, newHeader in headers.items()
                }
                for record in lod
            ]
        self.tables[name] = lod

    def hasTable(self, name: str):
        """
        Checks if table under given name exists

        Args:
            name(str): name of the Table

        Retruns:
            True if table exists otherwise False
        """
        return name in self.tables

    def saveToFile(self, fileName: str = None, dir_name: str = None):
        """
        saves SpreadSheet to file

        Args:
            fileName(str): name of the file if None SpreadSheet name is used
            dir_name(str): name of directory to store the file

        Returns:
            Nothing
        """
        if fileName is None:
            fileName = self.filename
        if dir_name is not None:
            fileName = os.path.join(dir_name, fileName)
        documentBuffer = self.toBytesIO()
        with open(fileName, "wb") as f:
            documentBuffer.seek(0)
            f.write(documentBuffer.read())

    def toBytesIO(self) -> BytesIO:
        """
        Converts the document into an BytesIO stream

        Returns:
            BytesIO Stream of the document
        """
        raise NotImplementedError

    def loadFromFile(self, file, samples: dict = None):
        """
        Load SpreadSheet from given file or file object
        """
        tables = self._loadFromFile(file)
        if tables:
            for name, table in tables.items():
                if samples:
                    if name in samples:
                        self.fixLodTypes(table, samples[name])
                self.tables[name] = table

    def _loadFromFile(self, file):
        """
        load the document from the given .ods file
        Args:
            file: absolut file path to the file that should be loaded
            samples(dict): samples of the sheets. Expected format: sheetName:SamplesForSheet
        Returns:

        """
        if isinstance(file, str):
            try:
                with open(file, mode="rb") as f:
                    buffer = BytesIO()
                    buffer.write(f.read())
                    # work around along the line of
                    # https://stackoverflow.com/a/42811024/1497139
                    buffer.name = f.name
            except Exception as e:
                print(f"Tried to open {file} as a File and failed")
                raise e
        else:
            buffer = file
        return self._loadFromBuffer(buffer)

    def _loadFromBuffer(self, buffer):
        """
        Load SpreadSheet from given buffer

        Args:
            buffer: file like object

        """
        raise NotImplementedError

    @property
    def filename(self):
        return self.name + self.FILE_TYPE

    @staticmethod
    def fixLodTypes(lod: list, samples: list, typeConversionMap: dict = None):
        """
        Fixes the types of the values of the given lod by converting it to the type corresponding to the given sampeles

        Args:
            lod(list): List of dicts to be type fixed
            samples(list): list of samples specifying the value types
            typeConversionMap(dict): Map from type to corresponding conversion function. If None default conversions for string values are used.
        """
        if typeConversionMap is None:

            def toDate(value):
                if isinstance(value, str):
                    return dateparser.parse(value).date()
                elif isinstance(value, datetime):
                    return value.date()
                elif isinstance(value, date):
                    return value
                else:
                    print(f"{value} could not be converted to date")
                    return value

            typeConversionMap = {
                str: lambda value: str(value),
                int: lambda value: int(value),
                float: lambda value: float(value),
                date: lambda value: toDate(value),
                datetime: lambda value: dateparser.parse(value),
            }
        # build sample types map
        sampleTypes = {}
        for sample in samples:
            if isinstance(sample, dict):
                for key, value in sample.items():
                    valueType = type(value)
                    if key not in sampleTypes:
                        sampleTypes[key] = valueType
                    elif sampleTypes[key] != valueType:
                        print(
                            f"Sample has inconsistent types for {key} the types {sampleTypes[key]} and {valueType} are defined"
                        )
                    else:
                        pass
        # fix types of lod
        for d in lod:
            if isinstance(d, dict):
                for key, value in d.items():
                    if (
                        value is not None
                        and key in sampleTypes
                        and sampleTypes[key] in typeConversionMap
                    ):
                        if type(value) != sampleTypes[key]:
                            d[key] = typeConversionMap[sampleTypes[key]](value)
            else:
                print("List of dicts contains a non dict item")

__init__(name, spreadSheetType)

constructor

Source code in spreadsheet/spreadsheet.py

def __init__(self, name: str, spreadSheetType: SpreadSheetType):
    """
    constructor
    """
    self.name = name
    self.spreadSheetType = spreadSheetType
    self.tables = {}  # dict of lods
    pass

addTable(name, lod, headers=None)

add the given data as table to the document

Parameters:

Name	Type	Description	Default
name(str)		name of the table	required
lod	list	data that should be added to the document as table	required
headers(dict)		Mapping from dict key to the new headers. Also functions as restriction. If not defined dict key are used as headers	required

Source code in spreadsheet/spreadsheet.py

def addTable(self, name: str, lod: list, headers: dict = None):
    """
    add the given data as table to the document

    Args:
        name(str): name of the table
        lod: data that should be added to the document as table
        headers(dict): Mapping from dict key to the new headers. Also functions as restriction. If not defined dict key are used as headers
    """
    if headers:
        lod = [
            {
                newHeader: record.get(oldHeader, None)
                for oldHeader, newHeader in headers.items()
            }
            for record in lod
        ]
    self.tables[name] = lod

create(spreadSheetType, name) classmethod

create a SpreadSheet of the given types

Parameters:

Name	Type	Description	Default
spreadSheetType(SpreadSheetType)		the type of onlinespreadsheet to create	required
name(str)		the name of the onlinespreadsheet	required

Source code in spreadsheet/spreadsheet.py

@classmethod
def create(cls, spreadSheetType: SpreadSheetType, name: str):
    """
    create a SpreadSheet of the given types

    Args:
        spreadSheetType(SpreadSheetType): the type of onlinespreadsheet to create
        name(str): the name of the onlinespreadsheet

    """
    spreadSheet = None
    if spreadSheetType == SpreadSheetType.EXCEL:
        spreadSheet = ExcelDocument(name=name)
    elif spreadSheetType == SpreadSheetType.ODS:
        spreadSheet = OdsDocument(name=name)
    elif spreadSheetType == SpreadSheetType.CSV:
        spreadSheet = CSVSpreadSheet(name=name)
    return spreadSheet

fixLodTypes(lod, samples, typeConversionMap=None) staticmethod

Fixes the types of the values of the given lod by converting it to the type corresponding to the given sampeles

Parameters:

Name	Type	Description	Default
lod(list)		List of dicts to be type fixed	required
samples(list)		list of samples specifying the value types	required
typeConversionMap(dict)		Map from type to corresponding conversion function. If None default conversions for string values are used.	required

Source code in spreadsheet/spreadsheet.py

@staticmethod
def fixLodTypes(lod: list, samples: list, typeConversionMap: dict = None):
    """
    Fixes the types of the values of the given lod by converting it to the type corresponding to the given sampeles

    Args:
        lod(list): List of dicts to be type fixed
        samples(list): list of samples specifying the value types
        typeConversionMap(dict): Map from type to corresponding conversion function. If None default conversions for string values are used.
    """
    if typeConversionMap is None:

        def toDate(value):
            if isinstance(value, str):
                return dateparser.parse(value).date()
            elif isinstance(value, datetime):
                return value.date()
            elif isinstance(value, date):
                return value
            else:
                print(f"{value} could not be converted to date")
                return value

        typeConversionMap = {
            str: lambda value: str(value),
            int: lambda value: int(value),
            float: lambda value: float(value),
            date: lambda value: toDate(value),
            datetime: lambda value: dateparser.parse(value),
        }
    # build sample types map
    sampleTypes = {}
    for sample in samples:
        if isinstance(sample, dict):
            for key, value in sample.items():
                valueType = type(value)
                if key not in sampleTypes:
                    sampleTypes[key] = valueType
                elif sampleTypes[key] != valueType:
                    print(
                        f"Sample has inconsistent types for {key} the types {sampleTypes[key]} and {valueType} are defined"
                    )
                else:
                    pass
    # fix types of lod
    for d in lod:
        if isinstance(d, dict):
            for key, value in d.items():
                if (
                    value is not None
                    and key in sampleTypes
                    and sampleTypes[key] in typeConversionMap
                ):
                    if type(value) != sampleTypes[key]:
                        d[key] = typeConversionMap[sampleTypes[key]](value)
        else:
            print("List of dicts contains a non dict item")

getTable(name)

returns the data corresponding to the given table name Args: name: name of the table

Returns:

Type	Description
	LoD

Source code in spreadsheet/spreadsheet.py

def getTable(self, name: str):
    """
    returns the data corresponding to the given table name
    Args:
        name: name of the table

    Returns:
        LoD
    """
    if name in self.tables:
        return self.tables[name]

hasTable(name)

Checks if table under given name exists

Parameters:

Name	Type	Description	Default
name(str)		name of the Table	required

Retruns

True if table exists otherwise False

Source code in spreadsheet/spreadsheet.py

def hasTable(self, name: str):
    """
    Checks if table under given name exists

    Args:
        name(str): name of the Table

    Retruns:
        True if table exists otherwise False
    """
    return name in self.tables

load(document) classmethod

Tries to load the given document as SpreadSheet Args: document: onlinespreadsheet document to load

Returns:

Type	Description
	SpreadSheet

Source code in spreadsheet/spreadsheet.py

@classmethod
def load(cls, document):
    """
    Tries to load the given document as SpreadSheet
    Args:
        document: onlinespreadsheet document to load

    Returns:
        SpreadSheet
    """
    documentSpreadSheetType = None
    documentName = ""
    spreadsheet = None
    # TODO - use SpreadSheeeType enum instead
    spreadSheetTypes = [OdsDocument, ExcelDocument, CSVSpreadSheet]
    # TODO Get rid of Werkzeug dependency ...
    # if isinstance(document, FileStorage):
    #    document.stream.seek(0)
    #    for spreadSheetType in spreadSheetTypes:
    #        if document.filename.endswith(spreadSheetType.FILE_TYPE):
    #            documentName=document.filename[:-len(spreadSheetType.FILE_TYPE)]
    #            documentSpreadSheetType=spreadSheetType
    #            break
    if (
        isinstance(document, io.BytesIO)
        or isinstance(document, io.StringIO)
        or isinstance(document, io.TextIOWrapper)
    ):
        document.seek(0)
        for spreadSheetType in spreadSheetTypes:
            if document.name.endswith(spreadSheetType.FILE_TYPE):
                documentName = document.name[: -len(spreadSheetType.FILE_TYPE)]
                documentSpreadSheetType = spreadSheetType
                break
    elif isinstance(document, str):
        try:
            buffer = None
            with open(document, "rb") as f:
                buffer = io.BytesIO(f.read())
                buffer.name = document
            return cls.load(buffer)
        except Exception as e:
            print(e)
            raise e
    else:
        print("Unable to load SpreadSheet")
        return None
    if documentSpreadSheetType:
        spreadsheet = documentSpreadSheetType(name=documentName)
        spreadsheet.loadFromFile(document)
    return spreadsheet

loadFromFile(file, samples=None)

Load SpreadSheet from given file or file object

Source code in spreadsheet/spreadsheet.py

def loadFromFile(self, file, samples: dict = None):
    """
    Load SpreadSheet from given file or file object
    """
    tables = self._loadFromFile(file)
    if tables:
        for name, table in tables.items():
            if samples:
                if name in samples:
                    self.fixLodTypes(table, samples[name])
            self.tables[name] = table

saveToFile(fileName=None, dir_name=None)

saves SpreadSheet to file

Parameters:

Name	Type	Description	Default
fileName(str)		name of the file if None SpreadSheet name is used	required
dir_name(str)		name of directory to store the file	required

Returns:

Type	Description
	Nothing

Source code in spreadsheet/spreadsheet.py

def saveToFile(self, fileName: str = None, dir_name: str = None):
    """
    saves SpreadSheet to file

    Args:
        fileName(str): name of the file if None SpreadSheet name is used
        dir_name(str): name of directory to store the file

    Returns:
        Nothing
    """
    if fileName is None:
        fileName = self.filename
    if dir_name is not None:
        fileName = os.path.join(dir_name, fileName)
    documentBuffer = self.toBytesIO()
    with open(fileName, "wb") as f:
        documentBuffer.seek(0)
        f.write(documentBuffer.read())

toBytesIO()

Converts the document into an BytesIO stream

Returns:

Type	Description
BytesIO	BytesIO Stream of the document

Source code in spreadsheet/spreadsheet.py

def toBytesIO(self) -> BytesIO:
    """
    Converts the document into an BytesIO stream

    Returns:
        BytesIO Stream of the document
    """
    raise NotImplementedError

SpreadSheetType

Bases: Enum

Entities of openresearch. Used to specify for a fixer the domain of operation.

Source code in spreadsheet/spreadsheet.py

class SpreadSheetType(Enum):
    """
    Entities of openresearch.
    Used to specify for a fixer the domain of operation.
    """

    CSV = auto()
    EXCEL = auto()
    ODS = auto()
    JSON = auto()

    def getProperty(self, propertyName):
        value = Format.formatMap[self.name][propertyName]
        return value

    def getPostfix(self):
        return self.getProperty("postfix")

    def getName(self):
        return self.getProperty("name")

    def getMimeType(self):
        return self.getProperty("mimetype")

    def getTitle(self):
        return self.getProperty("title")

    @classmethod
    def asSelectFieldChoices(cls):
        choices = []
        for i, choice in enumerate(cls):
            choices.append((choice.name, choice.getTitle()))
        return choices

tableediting

Created on 2021-12-08

@author: wf

TableEditing

Bases: object

table Editing

enhancement onlinespreadsheet editing call validation

Source code in spreadsheet/tableediting.py

class TableEditing(object):
    """
    table Editing

    enhancement
    onlinespreadsheet editing call
    validation
    """

    def __init__(self, lods: dict = None):
        """
        Constructor

        Args:
            lods(dict): a dict of list of dicts that represents the content of a Spreadsheet
        """
        if lods is None:
            self.lods = {}
        else:
            self.lods = lods
        self.enhanceCallbacks = []  # functions to call for enhancing

    def toSpreadSheet(self, spreadSheetType: SpreadSheetType, name: str) -> SpreadSheet:
        """
        convert me to the given spreadSheetType

        Args:
            spreadSheetType(SpreadSheetType): the type of onlinespreadsheet to create
            name(str): the name of the onlinespreadsheet
        """
        spreadSheet = SpreadSheet.create(spreadSheetType, name=name)
        spreadSheet.tables = self.lods
        return spreadSheet

    def fromSpreadSheet(self, spreadSheet: SpreadSheet):
        pass

    def addLoD(self, name, lod):
        """
        add the given list of dicts with the given name to my lods

        Args:
            name(str): the name
            lod(list): the list of dicts to add
        """
        self.lods[name] = lod

    def addEnhancer(self, callback):
        """
        add the given enhancer callback to my callbacks

        Args:
            callback(func): the callback function to add
        """
        self.enhanceCallbacks.append(callback)

    def enhance(self):
        """
        enhance/enrich my list of dicts with the set callbacks
        """
        for callback in self.enhanceCallbacks:
            callback(self)

__init__(lods=None)

Constructor

Parameters:

Name	Type	Description	Default
lods(dict)		a dict of list of dicts that represents the content of a Spreadsheet	required

Source code in spreadsheet/tableediting.py

def __init__(self, lods: dict = None):
    """
    Constructor

    Args:
        lods(dict): a dict of list of dicts that represents the content of a Spreadsheet
    """
    if lods is None:
        self.lods = {}
    else:
        self.lods = lods
    self.enhanceCallbacks = []  # functions to call for enhancing

addEnhancer(callback)

add the given enhancer callback to my callbacks

Parameters:

Name	Type	Description	Default
callback(func)		the callback function to add	required

Source code in spreadsheet/tableediting.py

def addEnhancer(self, callback):
    """
    add the given enhancer callback to my callbacks

    Args:
        callback(func): the callback function to add
    """
    self.enhanceCallbacks.append(callback)

addLoD(name, lod)

add the given list of dicts with the given name to my lods

Parameters:

Name	Type	Description	Default
name(str)		the name	required
lod(list)		the list of dicts to add	required

Source code in spreadsheet/tableediting.py

def addLoD(self, name, lod):
    """
    add the given list of dicts with the given name to my lods

    Args:
        name(str): the name
        lod(list): the list of dicts to add
    """
    self.lods[name] = lod

enhance()

enhance/enrich my list of dicts with the set callbacks

Source code in spreadsheet/tableediting.py

def enhance(self):
    """
    enhance/enrich my list of dicts with the set callbacks
    """
    for callback in self.enhanceCallbacks:
        callback(self)

toSpreadSheet(spreadSheetType, name)

convert me to the given spreadSheetType

Parameters:

Name	Type	Description	Default
spreadSheetType(SpreadSheetType)		the type of onlinespreadsheet to create	required
name(str)		the name of the onlinespreadsheet	required

Source code in spreadsheet/tableediting.py

def toSpreadSheet(self, spreadSheetType: SpreadSheetType, name: str) -> SpreadSheet:
    """
    convert me to the given spreadSheetType

    Args:
        spreadSheetType(SpreadSheetType): the type of onlinespreadsheet to create
        name(str): the name of the onlinespreadsheet
    """
    spreadSheet = SpreadSheet.create(spreadSheetType, name=name)
    spreadSheet.tables = self.lods
    return spreadSheet

version

Created on 2022-04-21

@author: wf

Version

Bases: object

Version handling for pyGenericSpreadSheet

Source code in spreadsheet/version.py

class Version(object):
    """
    Version handling for pyGenericSpreadSheet
    """

    version = spreadsheet.__version__
    date = "2022-04-21"
    updated = "2024-05-19"
    name = "pyGenericSpreadSheet"
    description = "python API providing generic Access to specific Spreadsheet backends"