Feature/write with blob

TM1py/Services/ApplicationService.py

@@ -229,7 +229,7 @@ def update(self, application: Union[Application, DocumentApplication], private:
         else:
             url = "/api/v1/Contents('Applications')" + mid + "/" + contents
             response = self._rest.POST(url, application.body, **kwargs)
-            
+
         return response
 
     def update_or_create_document_from_file(self, path: str, name: str,
@@ -245,10 +245,10 @@ def update_or_create_document_from_file(self, path: str, name: str,
 
         if self.exists(path=path, application_type=ApplicationTypes.DOCUMENT, name=name, private=private):
             response = self.update_document_from_file(path_to_file=path_to_file, application_path=path,
-                                                    application_name=name, private=private)
+                                                      application_name=name, private=private)
         else:
             response = self.create_document_from_file(path_to_file=path_to_file, application_path=path,
-                                                    application_name=name, private=private)
+                                                      application_name=name, private=private)
 
         return response
 
@@ -290,7 +290,7 @@ def create_document_from_file(self, path_to_file: str, application_path: str, ap
         """
         with open(path_to_file, 'rb') as file:
             application = DocumentApplication(path=application_path, name=application_name, content=file.read())
-            return self.create(application=application, private=private, **kwargs)            
+            return self.create(application=application, private=private, **kwargs)
 
     def update_document_from_file(self, path_to_file: str, application_path: str, application_name: str,
                                   private: bool = False, **kwargs) -> Response:

TM1py/Services/FileService.py

@@ -0,0 +1,66 @@
+# -*- coding: utf-8 -*-
+import json
+
+from TM1py.Services import RestService
+from TM1py.Services.ObjectService import ObjectService
+from TM1py.Utils import format_url
+
+
+class FileService(ObjectService):
+
+    def __init__(self, tm1_rest: RestService):
+        """
+
+        :param tm1_rest:
+        """
+        super().__init__(tm1_rest)
+        self._rest = tm1_rest
+
+    def get(self, file_name: str, **kwargs) -> bytes:
+        url = format_url(
+            "/api/v1/Contents('Blobs')/Contents('{name}')/Content",
+            name=file_name)
+
+        return self._rest.GET(url, **kwargs).content
+
+    def create(self, file_name: str, file_content: bytes, **kwargs):
+        url = "/api/v1/Contents('Blobs')/Contents"
+        body = {
+            "@odata.type": "#ibm.tm1.api.v1.Document",
+            "ID": file_name,
+            "Name": file_name
+        }
+        self._rest.POST(url, json.dumps(body), **kwargs)
+
+        url = format_url(
+            "/api/v1/Contents('Blobs')/Contents('{name}')/Content",
+            name=file_name)
+
+        return self._rest.PUT(url, file_content, headers=self.BINARY_HTTP_HEADER, **kwargs)
+
+    def update(self, file_name: str, file_content: bytes, **kwargs):
+        url = format_url(
+            "/api/v1/Contents('Blobs')/Contents('{name}')/Content",
+            name=file_name)
+
+        return self._rest.PUT(url, file_content, headers=self.BINARY_HTTP_HEADER, **kwargs)
+
+    def update_or_create(self, file_name: str, file_content: bytes, **kwargs):
+        if self.exists(file_name, **kwargs):
+            return self.update(file_name, file_content, **kwargs)
+
+        return self.create(file_name, file_content, **kwargs)
+
+    def exists(self, file_name: str, **kwargs):
+        url = format_url(
+            "/api/v1/Contents('Blobs')/Contents('{name}')",
+            name=file_name)
+
+        return self._exists(url, **kwargs)
+
+    def delete(self, file_name: str, **kwargs):
+        url = format_url(
+            "/api/v1/Contents('Blobs')/Contents('{name}')",
+            name=file_name)
+
+        return self._rest.DELETE(url, **kwargs)

TM1py/Services/TM1Service.py

@@ -3,6 +3,7 @@
 from TM1py.Services import HierarchyService, SecurityService, ApplicationService, SubsetService, ServerService, \
     MonitoringService, ProcessService, PowerBiService, AnnotationService, ViewService, RestService, CellService, \
     ChoreService, DimensionService, CubeService, ElementService, SandboxService, GitService
+from TM1py.Services.FileService import FileService
 
 
 class TM1Service:
@@ -31,6 +32,7 @@ def __init__(self, **kwargs):
         self.views = ViewService(self._tm1_rest)
         self.sandboxes = SandboxService(self._tm1_rest)
         self.git = GitService(self._tm1_rest)
+        self.files = FileService(self._tm1_rest)
 
     def logout(self, **kwargs):
         self._tm1_rest.logout(**kwargs)

Tests/FileService_test.py

@@ -0,0 +1,60 @@
+import configparser
+import unittest
+from pathlib import Path
+
+from TM1py import TM1Service
+from .Utils import skip_if_insufficient_version
+
+
+class TestApplicationService(unittest.TestCase):
+    tm1: TM1Service
+
+    FILE_NAME1 = "TM1py_unittest_file1"
+    FILE_NAME2 = "TM1py_unittest_file2"
+
+    @classmethod
+    def setUp(cls) -> None:
+        cls.config = configparser.ConfigParser()
+        cls.config.read(Path(__file__).parent.joinpath('config.ini'))
+        cls.tm1 = TM1Service(**cls.config['tm1srv01'])
+
+        with open(Path(__file__).parent.joinpath('resources', 'file.csv'), "rb") as file:
+            cls.tm1.files.update_or_create(cls.FILE_NAME1, file.read())
+
+        if cls.tm1.files.exists(cls.FILE_NAME2):
+            cls.tm1.files.delete(cls.FILE_NAME2)
+
+    @skip_if_insufficient_version(version="11.4")
+    def test_create_get(self):
+        with open(Path(__file__).parent.joinpath('resources', 'file.csv'), "rb") as original_file:
+            self.tm1.files.create(self.FILE_NAME2, original_file.read())
+
+            created_file = self.tm1.files.get(self.FILE_NAME1)
+
+        with open(Path(__file__).parent.joinpath('resources', 'file.csv'), "rb") as original_file:
+            self.assertEqual(original_file.read(), created_file)
+
+    @skip_if_insufficient_version(version="11.4")
+    def test_update_get(self):
+        with open(Path(__file__).parent.joinpath('resources', 'file.csv'), "rb") as original_file:
+            self.tm1.files.update(self.FILE_NAME1, original_file.read())
+
+            created_file = self.tm1.files.get(self.FILE_NAME1)
+
+        with open(Path(__file__).parent.joinpath('resources', 'file.csv'), "rb") as original_file:
+            self.assertEqual(original_file.read(), created_file)
+
+    @skip_if_insufficient_version(version="11.4")
+    def test_delete_exists(self):
+        self.assertTrue(self.tm1.files.exists(self.FILE_NAME1))
+
+        self.tm1.files.delete(self.FILE_NAME1)
+
+        self.assertFalse(self.tm1.files.exists(self.FILE_NAME1))
+
+    def tearDown(self) -> None:
+        if self.tm1.files.exists(self.FILE_NAME1):
+            self.tm1.files.delete(self.FILE_NAME1)
+        if self.tm1.files.exists(self.FILE_NAME2):
+            self.tm1.files.delete(self.FILE_NAME2)
+        self.tm1.logout()

Tests/resources/file.csv

@@ -0,0 +1,2 @@
+d1,d2,value
+e6,e2,888