Skip to content

Instantly share code, notes, and snippets.

@lethe2211
Last active November 8, 2022 00:38
Show Gist options
  • Save lethe2211/df9c5124170689ff5f3f0f974d601e7e to your computer and use it in GitHub Desktop.
Save lethe2211/df9c5124170689ff5f3f0f974d601e7e to your computer and use it in GitHub Desktop.
How to connect Azure Cosmos DB via REST API by curl https://docs.microsoft.com/en-us/rest/api/documentdb/
from wsgiref.handlers import format_date_time
from datetime import datetime
from time import mktime
import hmac
import hashlib
import base64
from urllib.parse import urlencode, quote_plus, quote
# Current HTTP date format like 'Thu, 27 Apr 2017 00:51:12 GMT'
date = format_date_time(mktime(datetime.now().timetuple()))
# print(date)
# Account info
account_id = '<account_id>' # Please input your account_id
base_url = 'https://{0}.documents.azure.com:443/'.format(account_id)
master_key = '<master_key>' # Please input your master_key. master_key is encoded by base64 and needs to be decoded during authentication
def generate_content_type():
'''
Generate Content-Type header (Required on PUT and POST)
'''
content_type = 'application/json' # Basically content_type is 'application/json'
# TODO:
# For POST on query operations, it must be application/query+json.
# For attachments, must be set to the Mime type of the attachment.
# https://docs.microsoft.com/ja-jp/rest/api/documentdb/common-documentdb-rest-request-headers
return 'Content-Type: {0}'.format(content_type)
def generate_x_ms_date(date):
'''
Generate x-ms-date header (Required)
'''
return 'x-ms-date: {0}'.format(date)
def generate_x_ms_version():
'''
Generate x-ms-version header (Required)
'''
api_version = '2017-02-22' # Latest API version
return 'x-ms-version: {0}'.format(api_version)
def generate_authorization(verb, resource_type, resource_id, date, master_key):
'''
Generate Authorization header (Required)
'''
type_of_token = 'master' # 'master' or 'resource' based on token type which you specified
token_version = '1.0' # '1.0'
verb = verb.lower() # All characters should be lower case
resource_type = resource_type.lower() # All characters should be lower case
date = date.lower() # All characters should be lower case
payload = '{0}\n{1}\n{2}\n{3}\n{4}\n'.format(verb, resource_type, resource_id, date, '')
# print(payload)
# master_key needs to be decoded by base64
key = base64.b64decode(master_key)
# print(key)
# HMAC-SHA256
hash = hmac.new(key, bytes(payload, 'utf-8'), hashlib.sha256)
# base64 decode
signature = base64.b64encode(hash.digest()).decode('utf-8')
# print(signature)
# URL encoding
header_value = quote('type={0}&ver={1}&sig={2}'.format(type_of_token, token_version, signature))
# print(header_value)
header = 'Authorization: {0}'.format(header_value)
# print(header)
return header
import sys
if len(sys.argv) < 5:
print('Usage: $ python cosmos_db.py "HTTP method" "resource_type (\'dbs\', \'colls\', \'docs\' etc)" "resource_id (\'dbs/dbname/colls/collname\' and so on)" "access_path (without \'/\' in its head)" "HTTP body (Only in case of POST/PUT. Basically JSON style)"')
print('')
print('Assume that we already have "ToDoList" database which has "test1" collection')
print(r'Sample1: $ python cosmos_db.py get docs dbs/ToDoList/colls/test/docs/1 dbs/ToDoList/colls/test/docs/1')
print(r'Sample2: $ python cosmos_db.py get colls dbs/ToDoList dbs/ToDoList/colls')
print(r'Sample3: $ python cosmos_db.py post colls dbs/ToDoList dbs/ToDoList/colls "{\\\"id\\\": \\\"1\\\", \\\"some_key\\\": \\\"some value\\\"}"')
print(r'Sample4: $ python cosmos_db.py post docs dbs/ToDoList/colls/test1 dbs/ToDoList/colls/test1/docs "{\\\"id\\\": \\\"1\\\", \\\"some_key\\\": \\\"some value\\\"}"')
print(r'Sample5: $ python cosmos_db.py put docs dbs/ToDoList/colls/test1/docs/1 dbs/ToDoList/colls/test1/docs/1 "{\\\"id\\\": \\\"1\\\", \\\"some_key\\\": \\\"some value to replace\\\"}"')
sys.exit(1)
verb = sys.argv[1] # HTTP method: 'get', 'post', 'put' etc... (All characters should be lower case during authentication)
resource_type = sys.argv[2] # Resource type you want to manipulate: 'dbs', 'colls', 'docs' etc... (All characters should be lower case during authentication)
resource_id = sys.argv[3] # Resource ID you want to manipulate 'dbs/dbname/colls/collname\' etc... (All characters should be lower case during authentication)
access_path = sys.argv[4] # Access path without '/' in its head
body = sys.argv[5] # HTTP body to send (Only in case of POST and PUT)
# Output a curl command to connect Cosmos DB
if verb.upper() == 'POST' or verb.upper() == 'PUT':
print('curl -v -X{0} -d"{1}" -H "{2}" -H "{3}" -H "{4}" -H "{5}" {6}{7}'.format(
verb.upper(),
body,
generate_x_ms_date(date),
generate_x_ms_version(),
generate_authorization(verb,resource_type, resource_id, date, master_key),
generate_content_type(),
base_url,
access_path)
)
else:
print('curl -v -X{0} -H "{1}" -H "{2}" -H "{3}" -H "{4}" {5}{6}'.format(
verb.upper(),
generate_x_ms_date(date),
generate_x_ms_version(),
generate_authorization(verb,resource_type, resource_id, date, master_key),
generate_content_type(),
base_url,
access_path)
)
@lethe2211
Copy link
Author

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment