SharePoint API: create taxonomy field

Author: vvgrem@gmail.com

examples/onedrive/upload_file.py

@@ -0,0 +1,31 @@
+import os
+
+import msal
+
+from office365.graph_client import GraphClient
+from settings import settings
+
+
+def acquire_token():
+    """
+    Acquire token (MSAL)
+    """
+    authority_url = 'https://login.microsoftonline.com/{0}'.format(settings.get('tenant'))
+    app = msal.ConfidentialClientApplication(
+        authority=authority_url,
+        client_id=settings.get('client_credentials').get('client_id'),
+        client_credential=settings.get('client_credentials').get('client_secret')
+    )
+    result = app.acquire_token_for_client(scopes=["https://graph.microsoft.com/.default"])
+    return result
+
+
+client = GraphClient(acquire_token)
+target_drive = client.users["jdoe@mediadev8.onmicrosoft.com"].drive
+
+local_path = "../../tests/data/SharePoint User Guide.docx"
+with open(local_path, 'rb') as f:
+    file_content = f.read()
+file_name = os.path.basename(local_path)
+target_file = target_drive.root.upload(file_name, file_content).execute_query()
+print(f"File {target_file.web_url} has been uploaded")

examples/sharepoint/files/upload_file.py

@@ -5,9 +5,9 @@
 from office365.runtime.auth.user_credential import UserCredential
 from office365.sharepoint.client_context import ClientContext
 
-ctx = ClientContext.connect_with_credentials(settings['url'],
-                                             UserCredential(settings['user_credentials']['username'],
-                                                            settings['user_credentials']['password']))
+credentials = UserCredential(settings.get('user_credentials').get('username'),
+                             settings.get('user_credentials').get('password'))
+ctx = ClientContext(settings.get('url')).with_credentials(credentials)
 
 path = "../../../tests/data/SharePoint User Guide.docx"
 with open(path, 'rb') as content_file:

examples/sharepoint/files/upload_large_file.py

@@ -1,3 +1,5 @@
+import os
+
 from settings import settings
 
 from office365.runtime.auth.user_credential import UserCredential
@@ -8,15 +10,23 @@ def print_upload_progress(offset):
     print("Uploaded '{0}' bytes...".format(offset))
 
 
-if __name__ == '__main__':
-    credentials = UserCredential(settings['user_credentials']['username'],
-                                 settings['user_credentials']['password'])
-    ctx = ClientContext(settings['url']).with_credentials(credentials)
+credentials = UserCredential(settings['user_credentials']['username'],
+                             settings['user_credentials']['password'])
+ctx = ClientContext(settings['url']).with_credentials(credentials)
+
+target_url = "/Shared Documents"
+target_folder = ctx.web.get_folder_by_server_relative_url(target_url)
+size_chunk = 1000000
+local_path = "../../../tests/data/big_buck_bunny.mp4"
+# local_path = "../../../tests/data/SharePoint User Guide.docx"
+file_size = os.path.getsize(local_path)
 
-    size_1Mb = 1000000
-    local_path = "../../../tests/data/big_buck_bunny.mp4"
-    target_url = "/Shared Documents"
-    result_file = ctx.web.get_folder_by_server_relative_url(target_url) \
-        .files.create_upload_session(local_path, size_1Mb, print_upload_progress)
-    ctx.execute_query()
-    print('File {0} has been uploaded successfully'.format(result_file.serverRelativeUrl))
+if file_size > size_chunk:
+    result_file = target_folder.files.create_upload_session(local_path, size_chunk, print_upload_progress)
+else:
+    with open(local_path, 'rb') as content_file:
+        file_content = content_file.read()
+    name = os.path.basename(local_path)
+    result_file = target_folder.upload_file(name, file_content)
+ctx.execute_query()
+print('File {0} has been uploaded successfully'.format(result_file.serverRelativeUrl))

office365/sharepoint/fields/field.py

@@ -12,7 +12,7 @@ def __init__(self, context, resource_path=None):
         super().__init__(context, resource_path)
 
     @staticmethod
-    def resolve_field_type(type_id):
+    def resolve_field_type(type_id_or_name):
         from office365.sharepoint.fields.field_calculated import FieldCalculated
         from office365.sharepoint.fields.field_choice import FieldChoice
         from office365.sharepoint.fields.field_computed import FieldComputed
@@ -24,7 +24,8 @@ def resolve_field_type(type_id):
         from office365.sharepoint.fields.field_text import FieldText
         from office365.sharepoint.fields.field_url import FieldUrl
         from office365.sharepoint.fields.field_user import FieldUser
-        field_types = {
+        from office365.sharepoint.taxonomy.taxonomyField import TaxonomyField
+        field_known_types = {
             FieldType.Text: FieldText,
             FieldType.Calculated: FieldCalculated,
             FieldType.Choice: FieldChoice,
@@ -37,7 +38,10 @@ def resolve_field_type(type_id):
             FieldType.Currency: FieldCurrency,
             FieldType.Note: FieldMultiLineText
         }
-        return field_types.get(type_id, Field)
+        if isinstance(type_id_or_name, int):
+            return field_known_types.get(type_id_or_name, Field)
+        else:
+            return TaxonomyField if type_id_or_name == "TaxonomyFieldType" else Field
 
     @staticmethod
     def create_field_from_type(context, field_type):
@@ -51,6 +55,7 @@ def set_show_in_display_form(self, flag):
         """
         qry = ServiceOperationQuery(self, "setShowInDisplayForm", [flag])
         self.context.add_query(qry)
+        return self
 
     def set_show_in_edit_form(self, flag):
         """Sets the value of the ShowInEditForm property for this fields.
@@ -88,4 +93,6 @@ def set_property(self, name, value, persist_changes=True):
                 "getById", [value], self._parent_collection.resource_path)
         if name == "FieldTypeKind":
             self.__class__ = self.resolve_field_type(value)
+        elif name == "TypeAsString" and self.properties.get('FieldTypeKind', 0) == 0:
+            self.__class__ = self.resolve_field_type(value)
         return self

office365/sharepoint/fields/field_collection.py

@@ -44,29 +44,47 @@ def create_taxonomy_field(self, name, ssp_id, term_set_id, anchor_id="00000000-0
             "web_id": web_id,
             "target_field": target_field
         }
-
         if field_id is None:
             field_params["field_id"] = str(uuid.uuid1())
 
-        from office365.sharepoint.lists.list import List
-        if isinstance(self._parent, List):
-            parent_list = self._parent
-
-            def _list_loaded():
-                field_params["web_id"] = parent_list.parentWeb.properties["Id"]
-                field_params["list_id"] = parent_list.properties["Id"]
-                self._create_taxonomy_field_query(**field_params)
-            self._parent.ensure_properties(["Id", "ParentWeb"], _list_loaded)
+        def _create_taxonomy_field_inner():
+            from office365.sharepoint.lists.list import List
+            if isinstance(self._parent, List):
+                parent_list = self._parent
+
+                def _list_loaded():
+                    field_params["web_id"] = parent_list.parentWeb.properties["Id"]
+                    field_params["list_id"] = parent_list.properties["Id"]
+                    self._build_taxonomy_field_query(**field_params)
+                self._parent.ensure_properties(["Id", "ParentWeb"], _list_loaded)
+            else:
+
+                def _web_loaded():
+                    field_params["web_id"] = self.context.web.properties["Id"]
+                    self._build_taxonomy_field_query(**field_params)
+                self.context.web.ensure_property("Id", _web_loaded)
+
+        if text_field_id is None:
+            text_field_name = f"{uuid.uuid4().hex}"
+            text_field_schema = f'''
+            <Field Type="Note" DisplayName="{name}_0" Hidden="TRUE" CanBeDeleted="TRUE" ShowInViewForms="FALSE"
+                   StaticName="{text_field_name}" Name="{text_field_name}">
+            </Field>
+            '''
+            text_field = self.create_field_as_xml(text_field_schema)
+
+            def _after_text_field_created(resp):
+                field_params["text_field_id"] = text_field.properties["Id"]
+                _create_taxonomy_field_inner()
+            self.context.after_execute(_after_text_field_created, True)
         else:
-            def _web_loaded():
-                field_params["web_id"] = self.context.web.properties["Id"]
-                self._create_taxonomy_field_query(**field_params)
-            self.context.web.ensure_property("Id", _web_loaded)
+            _create_taxonomy_field_inner()
+
         return target_field
 
-    def _create_taxonomy_field_query(self, name, ssp_id, term_set_id, anchor_id,
-                                     field_id=None, text_field_id=None, web_id=None, list_id=None,
-                                     target_field=None):
+    def _build_taxonomy_field_query(self, name, ssp_id, term_set_id, anchor_id,
+                                    field_id=None, text_field_id=None, web_id=None, list_id=None,
+                                    target_field=None):
         """
 
         :param str text_field_id: Text Field Id
@@ -199,6 +217,7 @@ def _create_field_as_xml_query(self, schema_xml, field):
         self.add_child(field)
         qry = ServiceOperationQuery(self, "CreateFieldAsXml", None, schema_xml, "parameters", field)
         self.context.add_query(qry)
+        return field
 
     def get_by_id(self, _id):
         """Gets the fields with the specified ID."""

office365/sharepoint/fields/field_creation_information.py

@@ -11,6 +11,8 @@ def __init__(self, title, field_type_kind, description=None):
         Represents metadata about fields creation.
 
         :type title: str
+        :type field_type_kind: int
+        :type description: str or None
         """
         super(FieldCreationInformation, self).__init__()
         self.Title = title

tests/sharepoint/test_field.py

@@ -17,9 +17,8 @@ def test_1_get_site_fields(self):
         self.assertGreater(len(site_fields), 0)
 
     def test_2_get_field(self):
-        title_field = self.client.site.rootWeb.fields.get_by_internal_name_or_title(self.target_field_name)
-        self.client.load(title_field)
-        self.client.execute_query()
+        title_field = self.client.site.rootWeb.fields.\
+            get_by_internal_name_or_title(self.target_field_name).get().execute_query()
         self.assertIsNotNone(title_field.internal_name)
         self.assertEqual(title_field.internal_name, self.target_field_name)
         self.assertIsInstance(title_field, FieldText)

tests/sharepoint/test_taxonomy.py

@@ -1,11 +1,12 @@
+from office365.sharepoint.taxonomy.taxonomyField import TaxonomyField
+
 from tests.sharepoint.sharepoint_case import SPTestCase
 
 from office365.sharepoint.fields.field import Field
 
 
 class TestSPTaxonomy(SPTestCase):
-
-    target_field = None   # type: Field
+    target_field = None  # type: Field
 
     @classmethod
     def setUpClass(cls):
@@ -15,19 +16,37 @@ def setUpClass(cls):
     def tearDownClass(cls):
         pass
 
-    def test1_create_taxonomy_field(self):
+    def test1_create_list_tax_field(self):
         ssp_id = "f02be691-d551-462f-aaae-e1c89168cd0b"
         term_set_id = "b49f64b3-4722-4336-9a5c-56c326b344d4"
         text_field_id = "cd790052-00e3-4317-a090-365cd85795b6"
         # list_id = "b9b8e2ef-6f9a-400b-a218-6a4899ea0121"
         # web_id = "cd5c86f3-261c-4d3a-8a91-34d6392cdfb9"
-        tax_field = self.client.web.default_document_library().fields\
+        tax_field = self.client.web.default_document_library().fields \
             .create_taxonomy_field(name="Category",
                                    ssp_id=ssp_id,
                                    term_set_id=term_set_id,
                                    text_field_id=text_field_id).execute_query()
         self.assertIsNotNone(tax_field.resource_path)
         self.__class__.target_field = tax_field
 
-    def test2_delete_taxonomy_field(self):
+    def test2_get_tax_field(self):
+        existing_field = self.__class__.target_field.get().execute_query()
+        self.assertTrue(existing_field.properties.get('TypeAsString'), 'TaxonomyFieldType')
+        self.assertIsInstance(existing_field, TaxonomyField)
+        self.assertIsNotNone(existing_field.properties.get('TextField'))
+        self.assertIsNotNone(existing_field.properties.get('LookupList'))
+        self.assertIsNotNone(existing_field.properties.get('LookupWebId'))
+
+        text_field_id = existing_field.properties.get('TextField')
+        text_field = self.client.web.default_document_library().fields.get_by_id(text_field_id)
+        self.client.load(text_field)
+        self.client.execute_batch()
+        self.assertIsNotNone(text_field.internal_name)
+
+    def test3_delete_tax_field(self):
         self.__class__.target_field.delete_object().execute_query()
+
+        # text_field_id = self.__class__.target_field.properties.get('TextField')
+        # text_field = self.client.web.default_document_library().fields.get_by_id(text_field_id)
+        # self.assertIsNotNone(text_field.resource_path)