WIP

Author: ppolewicz

b2sdk/raw_api.py

@@ -24,6 +24,7 @@
 from .b2http import B2Http
 from .exception import FileOrBucketNotFound, ResourceNotFound, UnusableFileName, InvalidMetadataDirective, WrongEncryptionModeForBucketDefault
 from .encryption.setting import EncryptionAlgorithm, EncryptionMode, EncryptionSetting
+from .file_lock import RetentionMode, RetentionPeriod, RetentionSetting
 from .utils import b2_url_encode, hex_sha1_of_stream
 
 # All possible capabilities
@@ -36,6 +37,12 @@
     'deleteBuckets',
     'readBucketEncryption',
     'writeBucketEncryption',
+    'readBucketRetentions',
+    'writeBucketRetentions',
+    'writeFileRetentions',
+    'writeFileLegalHolds',
+    'readFileRetentions',
+    'readFileLegalHolds',
     'listFiles',
     'readFiles',
     'shareFiles',
@@ -118,6 +125,7 @@ def create_bucket(
         cors_rules=None,
         lifecycle_rules=None,
         default_server_side_encryption: Optional[EncryptionSetting] = None,
+        is_file_lock_enabled: Optional[bool] = None,
     ):
         pass
 
@@ -272,6 +280,7 @@ def update_bucket(
         lifecycle_rules=None,
         if_revision_is=None,
         default_server_side_encryption: Optional[EncryptionSetting] = None,
+        default_retention: Optional[RetentionSetting] = None,
     ):
         pass
 
@@ -345,6 +354,16 @@ def _post_json(self, base_url, api_name, auth, **params) -> Dict[str, Any]:
         headers = {'Authorization': auth}
         return self.b2_http.post_json_return_json(url, headers, params)
 
+    def _get_json(self, base_url, api_name, auth, **params) -> Dict[str, Any]:
+        # this is useful for testing purposes only
+        url = '%s/b2api/%s/%s' % (base_url, API_VERSION, api_name)
+        import urllib
+        url += '?' + urllib.parse.urlencode(params)
+        headers = {'Authorization': auth}
+        print('eee')
+        with self.b2_http.get_content(url, headers) as response:
+            return response.json()
+
     def authorize_account(self, realm_url, application_key_id, application_key):
         auth = b'Basic ' + base64.b64encode(
             ('%s:%s' % (application_key_id, application_key)).encode()
@@ -364,7 +383,8 @@ def create_bucket(
         bucket_info=None,
         cors_rules=None,
         lifecycle_rules=None,
-        default_server_side_encryption=None,
+        default_server_side_encryption: Optional[EncryptionSetting] = None,
+        is_file_lock_enabled: Optional[bool] = None,
     ):
         kwargs = dict(
             accountId=account_id,
@@ -382,6 +402,10 @@ def create_bucket(
                 raise WrongEncryptionModeForBucketDefault(default_server_side_encryption.mode)
             kwargs['defaultServerSideEncryption'
                   ] = default_server_side_encryption.serialize_to_json_for_request()
+        if is_file_lock_enabled is not None:
+            #kwargs['isFileLockEnabled'] = is_file_lock_enabled
+            kwargs['isFileLockEnabled'] = is_file_lock_enabled and "true" or "false"
+        #return self._get_json(
         return self._post_json(
             api_url,
             'b2_create_bucket',
@@ -653,7 +677,8 @@ def update_bucket(
         cors_rules=None,
         lifecycle_rules=None,
         if_revision_is=None,
-        default_server_side_encryption=None,
+        default_server_side_encryption: Optional[EncryptionSetting] = None,
+        default_retention: Optional[RetentionSetting] = None,
     ):
         assert bucket_info is not None or bucket_type is not None
 
@@ -669,11 +694,12 @@ def update_bucket(
         if lifecycle_rules is not None:
             kwargs['lifecycleRules'] = lifecycle_rules
         if default_server_side_encryption is not None:
-            if default_server_side_encryption is not None:
-                if not default_server_side_encryption.mode.can_be_set_as_bucket_default():
-                    raise WrongEncryptionModeForBucketDefault(default_server_side_encryption.mode)
-                kwargs['defaultServerSideEncryption'
-                      ] = default_server_side_encryption.serialize_to_json_for_request()
+            if not default_server_side_encryption.mode.can_be_set_as_bucket_default():
+                raise WrongEncryptionModeForBucketDefault(default_server_side_encryption.mode)
+            kwargs['defaultServerSideEncryption'
+                  ] = default_server_side_encryption.serialize_to_json_for_request()
+        if default_retention is not None:
+            kwargs['defaultRetention'] = default_retention.serialize_to_json_for_request()
 
         return self._post_json(
             api_url,
@@ -979,8 +1005,18 @@ def test_raw_api_helper(raw_api):
     bucket_name = 'test-raw-api-%s-%d-%d' % (
         account_id, int(time.time()), random.randint(1000, 9999)
     )
+
+    import http.client
+    http.client.HTTPConnection.debuglevel = 1
+
     bucket_dict = raw_api.create_bucket(
-        api_url, account_auth_token, account_id, bucket_name, 'allPublic'
+        api_url,
+        account_auth_token,
+        account_id,
+        bucket_name,
+        'allPublic',
+        is_file_lock_enabled=False,
+        #is_file_lock_enabled=True,
     )
     bucket_id = bucket_dict['bucketId']
     first_bucket_revision = bucket_dict['revision']
@@ -1186,7 +1222,10 @@ def test_raw_api_helper(raw_api):
         account_id,
         bucket_id,
         'allPrivate',
-        bucket_info={'color': 'blue'}
+        bucket_info={'color': 'blue'},
+        default_retention=RetentionSetting(
+            mode=RetentionMode.GOVERNANCE, period=RetentionPeriod(days=1)
+        ),
     )
     assert first_bucket_revision < updated_bucket['revision']