refactored GDocs

added Temp__GDocs__File

.github/workflows/ci-pipeline__dev.yml

@@ -41,7 +41,7 @@ jobs:
       - run-unit-tests
 
   publish-to-pypi:
-    #if: False
+    if: False
     name: "Publish to: PYPI"
     permissions:
       id-token: write

osbot_gsuite/gsuite/GSuite.py

@@ -48,7 +48,10 @@ def calendar_v3(self):
         return self.create_service('calendar','v3','calendar')
 
     def docs_v1(self):
-        return self.create_service('docs', 'v1', 'documents')
+        kwargs = dict(service_name = 'docs',
+                      version      = 'v1',
+                      scopes       = ['https://www.googleapis.com/auth/documents'])
+        return self.create_service(**kwargs)
 
     def drive_v3(self):
         kwargs = dict(service_name = 'drive',

osbot_gsuite/gsuite/docs/GDocs.py

@@ -0,0 +1,58 @@
+from googleapiclient.discovery import Resource
+from googleapiclient.errors import HttpError
+
+from osbot_utils.utils.Objects import dict_to_obj
+
+from osbot_utils.utils.Dev import pprint
+
+from osbot_utils.decorators.methods.cache_on_self import cache_on_self
+
+from osbot_gsuite.gsuite.GSuite import GSuite
+from osbot_gsuite.gsuite.drive.GDrive import GDrive
+from osbot_utils.base_classes.Type_Safe import Type_Safe
+
+
+class GDocs(Type_Safe):
+    gsuite : GSuite
+    gdrive : GDrive
+
+    # def __init__(self, gsuite_secret_id=None):
+    #     self.docs   = GSuite(gsuite_secret_id).docs_v1().documents()
+    #     self.gdrive = GDrive(gsuite_secret_id)
+
+    @cache_on_self
+    def docs_v1(self) -> Resource:
+        return self.gsuite.docs_v1()
+
+    def document_create(self, title, folder_id=None):
+        body     = { 'title': title }
+        doc_info = self.documents().create(body=body).execute()
+        doc_id   = doc_info.get('documentId')
+        if folder_id:
+            self.gdrive.file_move_to_folder(doc_id, folder_id)
+
+        return dict_to_obj(dict(doc_info  = doc_info ,
+                                doc_id    = doc_id   ,
+                                folder_id = folder_id))
+
+    def document_delete(self, doc_id):
+        return self.gdrive.file_delete(doc_id)
+
+    def document_exists(self, doc_id):
+        return self.document_info(doc_id) is not None
+
+    def document_info(self, doc_id):
+        try:
+            doc_info = self.documents().get(documentId=doc_id).execute()
+            return dict_to_obj(doc_info)
+        except HttpError as http_error:
+            if http_error.status_code != 404:
+                raise http_error
+
+    @cache_on_self
+    def documents(self) -> Resource:
+        return self.docs_v1().documents()
+
+    def execute_requests(self, file_id, requests):
+        body = {'requests': requests}
+        return self.documents().batchUpdate(documentId= file_id, body=body).execute()

osbot_gsuite/gsuite/docs/Temp__GDocs__File.py

@@ -0,0 +1,38 @@
+from types                                          import SimpleNamespace
+from osbot_gsuite.gsuite.docs.GDocs                 import GDocs
+from osbot_utils.utils.Misc                         import random_text
+from osbot_gsuite.testing.OSBot__GSuite__Testing    import osbot_gsuite_testing
+from osbot_utils.base_classes.Type_Safe             import Type_Safe
+
+
+class Temp__GDocs__File(Type_Safe):
+    gdocs    : GDocs
+    title    : str = random_text('osbot-gsuite-random-file')
+    doc_data : SimpleNamespace
+    doc_id   : str
+
+    def parent_folder(self):
+        return osbot_gsuite_testing.gdrive_temp_folder()
+
+    def __enter__(self):
+        self.file_create()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.file_delete()
+
+    def file_create(self):
+        kwargs = dict(title   = self.title    ,
+                      folder_id = self.parent_folder())
+        self.doc_data = self.gdocs.document_create(**kwargs)
+        self.doc_id   = self.doc_data.doc_id
+        return self.doc_id
+
+    def file_delete(self):
+        return self.gdocs.document_delete(self.doc_id)
+
+    def file_exists(self):
+        return self.gdocs.document_exists(self.doc_id)
+
+    def file_info(self):
+        return self.gdocs.document_info(self.doc_id)

osbot_gsuite/gsuite/drive/GDrive.py

@@ -74,6 +74,9 @@ def file_create(self, file_type, title, folder=None):
     def file_export(self, file_Id):
         return self.files().export(fileId=file_Id, mimeType='application/pdf').execute()
 
+    def file_move_to_folder(self, file_id, folder_id):
+        return self.files().update(fileId=file_id, addParents=folder_id, fields='id, parents').execute()
+
     def file_export_as_pdf_to(self,file_id,target_file):
         pdf_data = self.file_export(file_id)
         with open(target_file, "wb") as fh:

tests/integration/gsuite/docs/test_GDocs.py

@@ -0,0 +1,37 @@
+from unittest import TestCase
+
+from googleapiclient.discovery import Resource
+from osbot_utils.utils.Misc import random_text
+
+from osbot_utils.utils.Dev import pprint
+
+from osbot_gsuite.testing.OSBot__GSuite__Testing import osbot_gsuite_testing
+from osbot_utils.utils.Env import load_dotenv
+
+from osbot_gsuite.gsuite.docs.GDocs import GDocs
+
+
+class test_GDocs(TestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        cls.temp_folder_id = osbot_gsuite_testing.gdrive_temp_folder()
+        cls.gdocs = GDocs()
+
+    def test_setUpClass(self):
+        with self.gdocs as _:
+            assert type(_) is GDocs
+            assert type(_.docs_v1  ()) is Resource
+            assert type(_.documents()) is Resource
+
+    def test_document_create(self):
+        with self.gdocs as _:
+            temp_title = random_text('an temp title')
+            result     = _.document_create(title=temp_title, folder_id=self.temp_folder_id)
+            doc_id     = result.doc_id
+            doc_info   = result.doc_info
+            assert doc_info.title            == temp_title
+            assert doc_info.documentId       == doc_id
+            assert _.document_delete(doc_id) is True
+
+

tests/integration/gsuite/docs/test_Temp__GDoc__File.py

@@ -0,0 +1,37 @@
+from unittest                                       import TestCase
+
+from osbot_utils.utils.Misc import wait_for
+
+from osbot_utils.utils.Dev import pprint
+
+from osbot_gsuite.gsuite.docs.Temp__GDocs__File import Temp__GDocs__File
+from osbot_gsuite.testing.OSBot__GSuite__Testing    import ENV_NAME__GDRIVE__TEMP_FOLDER
+from osbot_utils.utils.Env                          import load_dotenv, get_env
+
+
+
+class test_Temp__GDocs__File(TestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        load_dotenv()
+        cls.temp_gdocs_files = Temp__GDocs__File()
+
+    def test_parent_folder(self):
+        assert self.temp_gdocs_files.parent_folder() == get_env(ENV_NAME__GDRIVE__TEMP_FOLDER)
+
+    def test__enter__exit__(self):
+        with self.temp_gdocs_files as _:
+            file_info = _.file_info()
+            assert _.file_exists()                                          is True
+            assert file_info.title                                             == _.title
+            assert file_info.documentId                                       == _.doc_id
+
+        #assert _.file_exists() is False                        # BUG: this is failing (the file is not being deleted immediately)
+        for i in range(20):                                     # interesting race condition here where the file can take a few seconds to be deleted
+            result = _.file_exists()
+            if result is False:
+                break
+            wait_for(0.2)                                       # it can take up to 4 secs
+
+        assert _.file_exists() is False                         # now we can confirm that the file has been deleted

tests/integration/gsuite/drive/test_Temp_GDrive_Folder.py

@@ -1,8 +1,6 @@
-from unittest import TestCase
-
-from osbot_gsuite.testing.OSBot__GSuite__Testing import ENV_NAME__GDRIVE__TEMP_FOLDER
-from osbot_utils.utils.Env import load_dotenv, get_env
-from osbot_utils.utils.Dev                          import pprint
+from unittest                                       import TestCase
+from osbot_gsuite.testing.OSBot__GSuite__Testing    import ENV_NAME__GDRIVE__TEMP_FOLDER
+from osbot_utils.utils.Env                          import load_dotenv, get_env
 from osbot_gsuite.gsuite.drive.Temp__GDrive__Folder import Temp__GDrive__Folder