User-defined Headers (SDK for Python)

Function

When calling an API, you can configure user-defined headers to meet specific needs. The SDK will automatically calculate the signature for the specified headers if needed.

Method:

You can add the specified headers in extensionHeaders in the dictionary format.

extensionHeaders is usually the last parameter of an API. To prevent parameter misplacement, you are advised to pass arguments using explicit parameter names, for example, extensionHeaders=extensionHeaders.

Restrictions

The mapping between OBS regions and endpoints must comply with what is listed in Regions and Endpoints.

Sample Code 1: Single-Connection Bandwidth Throttling

This example configures a user-defined header to limit the single-connection rate to 100 KB/s for downloading objectname from examplebucket.

     import os
import traceback
from obs import ObsClient

# Obtain an AK/SK pair using environment variables (recommended) or import it in other ways. Using hard coding may result in leakage.
# Obtain an AK and SK pair on the management console. For details, see https://support.huaweicloud.com/intl/en-us/usermanual-ca/ca_01_0003.html.
ak = os.getenv('AccessKeyID')
sk = os.getenv('SecretAccessKey')
# If you use a temporary AK/SK pair and a security token to access OBS, obtain them from environment variables.
# security_token = os.getenv('SecurityToken')
# Set server to the endpoint corresponding to the bucket. CN-Hong Kong is used here as an example. Replace it with the one currently in use.
server = 'https://obs.ap-southeast-1.myhuaweicloud.com'  
# Create an obsClient instance.
# If you use a temporary AK/SK pair and a security token to access OBS, you must specify security_token when creating an instance.
obs_client = ObsClient(access_key_id=ak, secret_access_key=sk, server=server) 
try: 
    bucket_name = 'examplebucket' 
    object_key = 'objectname' 

    # Configure the download rate limit by specifying x-obs-traffic-limit, in bits. The value range is from 819200 (100 KB/s) to 838860800 (100 MB/s). 819200 is used as an example.
    extension_headers = {'x-obs-traffic-limit': 819200} 
    # Specify the full path to which the object is downloaded. The full path contains the local file name.
    download_path = 'localfile' 
    # Download the object at a limited rate.
    resp = obs_client.getObject(bucket_name, object_key, download_path, extensionHeaders=extension_headers) 

    # If status code 2xx is returned, the API call succeeds. Otherwise, the API call fails.
    if resp.status < 300: 
        print('Get Object Succeeded') 
        print('requestId:', resp.requestId) 
        print('url:', resp.body.url) 
    else: 
        print('Get Object Failed') 
        print('requestId:', resp.requestId) 
        print('errorCode:', resp.errorCode) 
        print('errorMessage:', resp.errorMessage) 
except Exception: 
    print('Get Object Failed') 
    print(traceback.format_exc())
 
 
  

Sample Code 2: Upload Callback

This example configures a user-defined header to implement the callback for a common upload.

     import os
import json
import base64
import traceback
from urllib.parse import quote
from obs import ObsClient

# Obtain an AK/SK pair using environment variables (recommended) or import it in other ways. Using hard coding may result in leakage.
# Obtain an AK and SK pair on the management console. For details, see https://support.huaweicloud.com/intl/en-us/usermanual-ca/ca_01_0003.html.
ak = os.getenv('AccessKeyID')
sk = os.getenv('SecretAccessKey')
# If you use a temporary AK/SK pair and a security token to access OBS, obtain them from environment variables.
# security_token = os.getenv('SecurityToken')
# Set server to the endpoint corresponding to the bucket. CN-Hong Kong is used here as an example. Replace it with the one currently in use.
server = 'https://obs.ap-southeast-1.myhuaweicloud.com' 
# Create an obsClient instance.
# If you use a temporary AK/SK pair and a security token to access OBS, you must specify security_token when creating an instance.
obs_client = ObsClient(access_key_id=ak, secret_access_key=sk, server=server)
try:
    # Specify a protocol.
    protocol = 'http://'

    # Specify the callback address. If the URL contains any special characters or CJK characters, they must be URL-encoded using quote(str).
    callback_url1 = protocol + quote('www.example.com/callback1')
    callback_url2 = protocol + quote('www.example.com/CJK characters?key=Name in CJK characters")
    # (Optional) Specify the value of the host header included in the callback request. If this parameter is not specified, the value of host parsed from callbackUrl is used.
    callback_host = 'www.example.com'
    # Specify the body of the callback request.
    callback_body = 'key = $(key)&override = $(override)&size = $(size)&bucket = $(bucket)&etag = $(etag)'
    # Configure the upload callback.
    call_back_policy = {'callbackBody': callback_body, 'callbackUrl': callback_url1 + ';' + callback_url2,
                      'callbackHost': callback_host}

    # Configure the custom headers by specifying extensionHeaders. The input parameters are in the dictionary format.
    # Convert the upload callback configuration to a JSON string, then to a binary string (through json.dumps().encode()), and then encode the binary string using Base64 (through base64.b64encode()). The Base64-encoded data is of the bytes type and needs to be converted to str using bytes_data.decode('utf-8').
    extension_headers = {'x-obs-callback': base64.b64encode(json.dumps(call_back_policy).encode()).decode('utf-8')}

    bucket_name = 'your-bucket_name'
    object_key = 'example.txt'
    content = 'Hello OBS'
    # Upload the text and perform the upload callback.
    resp = obs_client.putContent(bucket_name, object_key, content, extensionHeaders=extension_headers)

    # If status code 2xx is returned, the API call succeeds. Otherwise, the API call fails.
    if resp.status < 300:
        print('Put Content Succeeded')
        print('requestId:', resp.requestId)
        print('etag:', resp.body.etag)
    else:
        print('Put Content Failed')
        print('requestId:', resp.requestId)
        print('errorCode:', resp.errorCode)
        print('errorMessage:', resp.errorMessage)
except Exception:
    print('Put Content Failed')
    print(traceback.format_exc())
 
 
  

This example configures a user-defined header to implement the callback for a resumable upload.

     import os
import json
import base64
import traceback
from urllib.parse import quote
from obs import ObsClient

# Obtain an AK/SK pair using environment variables (recommended) or import it in other ways. Using hard coding may result in leakage.
# Obtain an AK and SK pair on the management console. For details, see https://support.huaweicloud.com/intl/en-us/usermanual-ca/ca_01_0003.html.
ak = os.getenv('AccessKeyID')
sk = os.getenv('SecretAccessKey')
# If you use a temporary AK/SK pair and a security token to access OBS, obtain them from environment variables.
# security_token = os.getenv('SecurityToken')
# Set server to the endpoint corresponding to the bucket. CN-Hong Kong is used here as an example. Replace it with the one currently in use.
server = 'https://obs.ap-southeast-1.myhuaweicloud.com' 
# Create an obsClient instance.
# If you use a temporary AK/SK pair and a security token to access OBS, you must specify security_token when creating an instance.
obs_client = ObsClient(access_key_id=ak, secret_access_key=sk, server=server)
try:
    # Specify a protocol.
    protocol = 'http://'

    # Specify the callback address. If the URL contains any special characters or CJK characters, they must be URL-encoded using quote(str).
    callback_url1 = protocol + quote('www.example.com/callback1')
    callback_url2 = protocol + quote('www.example.com/CJK characters?key=Name in CJK characters")
    # (Optional) Specify the value of the host header included in the callback request. If this parameter is not specified, the value of host parsed from callbackUrl is used.
    callback_host = 'www.example.com'
    # Specify the body of the callback request.
    callback_body = 'key = $(key)&override = $(override)&size = $(size)&bucket = $(bucket)&etag = $(etag)'
    # Configure the upload callback.
    call_back_policy = {'callbackBody': callback_body, 'callbackUrl': callback_url1 + ';' + callback_url2,
                      'callbackHost': callback_host}

    # Configure the custom headers by specifying extensionHeaders. The input parameters are in the dictionary format.
    # Convert the upload callback configuration to a JSON string, then to a binary string (through json.dumps().encode()), and then encode the binary string using Base64 (through base64.b64encode()). The Base64-encoded data is of the bytes type and needs to be converted to str using bytes_data.decode('utf-8').
    extension_headers = {'x-obs-callback': base64.b64encode(json.dumps(call_back_policy).encode()).decode('utf-8')}

    bucket_name = 'your-bucket_name'
    object_key = 'example.txt'
    # Path of the local file to be uploaded in resumable mode. Replace it with the actual file path.
    file_path = './example.txt'
    # Set a callback for a resumable upload.
    resp = obs_client.uploadFile(bucket_name, object_key, file_path, extensionHeaders=extension_headers)

    # If status code 2xx is returned, the API call succeeds. Otherwise, the API call fails.
    if resp.status < 300:
        print('UploadFile Succeeded')
        print('requestId:', resp.requestId)
    else:
        print('UploadFile Failed')
        print('requestId:', resp.requestId)
        print('errorCode:', resp.errorCode)
        print('errorMessage:', resp.errorMessage)
except Exception:
    print('UploadFile Failed')
    print(traceback.format_exc())
 
 
  

Sample Code 3: Changing the Expiration Time of an Object

This example changes the expiration time of an object using a user-defined header.

     import os
import traceback
from obs import ObsClient

# Obtain an AK/SK pair using environment variables (recommended) or import it in other ways. Using hard coding may result in leakage.
# Obtain an AK and SK pair on the management console. For details, see https://support.huaweicloud.com/intl/en-us/usermanual-ca/ca_01_0003.html.
ak = os.getenv('AccessKeyID')
sk = os.getenv('SecretAccessKey')
# If you use a temporary AK/SK pair and a security token to access OBS, obtain them from environment variables.
# security_token = os.getenv('SecurityToken')
# Set server to the endpoint corresponding to the bucket. CN-Hong Kong is used here as an example. Replace it with the one currently in use.
server = 'https://obs.ap-southeast-1.myhuaweicloud.com'  
# Create an obsClient instance.
# If you use a temporary AK/SK pair and a security token to access OBS, you must specify security_token when creating an instance.
obs_client = ObsClient(access_key_id=ak, secret_access_key=sk, server=server) 
try: 
    bucket_name = 'examplebucket' 
    object_key = 'objectname' 

    # x-obs-expires indicates how many days after the last modification the object expires. This example configures 3 days.
    extensionHeaders = {'x-obs-expires': 3} 
    # Configure the object metadata.
    resp = obs_client.setObjectMetadata(bucket_name, object_key, extensionHeaders=extensionHeaders) 

    # If status code 2xx is returned, the API call succeeds. Otherwise, the API call fails.
    if resp.status < 300: 
        print('Set Object Metadata Succeeded') 
        print('requestId:', resp.requestId) 
    else: 
        print('Set Object Metadata Failed') 
        print('requestId:', resp.requestId) 
        print('errorCode:', resp.errorCode) 
        print('errorMessage:', resp.errorMessage) 
except Exception: 
    print('Set Object Metadata Failed') 
    print(traceback.format_exc())
 
 
  

Sample Code 4: Enabling WORM When Creating a Bucket

This example configures a user-defined header to enable WORM when creating a bucket.

     import os
import traceback
from obs import ObsClient, CreateBucketHeader

# Obtain an AK/SK pair using environment variables (recommended) or import it in other ways. Using hard coding may result in leakage.
# Obtain an AK and SK pair on the management console. For details, see https://support.huaweicloud.com/intl/en-us/usermanual-ca/ca_01_0003.html.
ak = os.getenv('AccessKeyID')
sk = os.getenv('SecretAccessKey')
# If you use a temporary AK/SK pair and a security token to access OBS, obtain them from environment variables.
# security_token = os.getenv('SecurityToken')
# Set server to the endpoint corresponding to the bucket. CN-Hong Kong is used here as an example. Replace it with the one currently in use.
server = 'https://obs.ap-southeast-1.myhuaweicloud.com'  
# Create an obsClient instance.
# If you use a temporary AK/SK pair and a security token to access OBS, you must specify security_token when creating an instance.
obs_client = ObsClient(access_key_id=ak, secret_access_key=sk, server=server) 
try: 
    # Specify the additional headers to create a private bucket that is in the Standard storage class.
    header = CreateBucketHeader(aclControl='PRIVATE', storageClass='STANDARD') 
    # Set location to the region corresponding to the bucket. Here uses CN North-Beijing4 as an example. Replace it with the one in use.
    location = 'cn-north-4' 
    bucket_name = 'examplebucket' 
    extensionHeaders = {'x-obs-bucket-object-lock-enabled' : 'true'} 
    # Create a bucket.
    resp = obs_client.createBucket(bucket_name, header, location, extensionHeaders) 

    # If status code 2xx is returned, the API call succeeds. Otherwise, the API call fails.
    if resp.status < 300: 
        print('Create Bucket Succeeded') 
        print('requestId:', resp.requestId) 
    else: 
        print('Create Bucket Failed') 
        print('requestId:', resp.requestId) 
        print('errorCode:', resp.errorCode) 
        print('errorMessage:', resp.errorMessage) 
except Exception: 
    print('Create Bucket Failed') 
    print(traceback.format_exc())
 
 
  

Helpful Links

OBS Error Codes

Parent topic: Other APIs (SDK for Python)

Previous topic: Generating Authentication Parameters for Browser-Based Uploads (SDK for Python)

Next topic: Data Types

Feedback

Was this page helpful?

Helpful Not helpful

Provide feedback

Thank you very much for your feedback. We will continue working to improve the documentation.See the reply and handling status in My Cloud VOC.

The system is busy. Please try again later.

For any further questions, feel free to contact us through the chatbot.

Chatbot