1
+ # -------------------------------------------------------------------------
2
+ # Copyright (c) Microsoft Corporation. All rights reserved.
3
+ # Licensed under the MIT License. See LICENSE.txt in the project root for
4
+ # license information.
5
+ # -------------------------------------------------------------------------
6
+ import json
7
+ import os
8
+ import sys
9
+ import traceback
10
+ import uuid
11
+
12
+ from azure .identity import DefaultAzureCredential , InteractiveBrowserCredential
13
+ from azure .cosmos import CosmosClient
14
+ import azure .cosmos .exceptions as exceptions
15
+ from azure .cosmos .partition_key import PartitionKey
16
+ import config
17
+
18
+ # ----------------------------------------------------------------------------------------------------------
19
+ # Prerequisites -
20
+ #
21
+ # 1. An Azure Cosmos account in fabric environment and database and container created.
22
+ # https://learn.microsoft.com/en-us/fabric/database/cosmos-db/overview
23
+ # 2. Python packages (preview + identity) and login:
24
+ # pip install "azure-cosmos==4.14.0b3" azure-identity
25
+ # az login
26
+ # ----------------------------------------------------------------------------------------------------------
27
+ # Sample - demonstrates how to authenticate and use your database account using AAD credentials with Fabric.
28
+ # Read more about operations allowed for this authorization method: https://aka.ms/cosmos-native-rbac
29
+ # ----------------------------------------------------------------------------------------------------------
30
+ # Note:
31
+ # This sample assumes the database and container already exist.
32
+ # It writes one item (PK path assumed to be "/pk") and reads it back.
33
+ # ----------------------------------------------------------------------------------------------------------
34
+ HOST = config .settings ["host" ]
35
+ DATABASE_ID = config .settings ["database_id" ]
36
+ CONTAINER_ID = config .settings ["container_id" ]
37
+ PARTITION_KEY = PartitionKey (path = "/pk" )
38
+
39
+ def get_test_item (num : int ) -> dict :
40
+ return {
41
+ "id" : f"Item_{ num } " ,
42
+ "pk" : "partition1" ,
43
+ "name" : "Item 1" ,
44
+ "description" : "This is item 1" ,
45
+ "runId" : str (uuid .uuid4 ())
46
+ }
47
+
48
+
49
+ def run_sample ():
50
+ # if you want to override scope for AAD authentication.
51
+ #os.environ["AZURE_COSMOS_AAD_SCOPE_OVERRIDE"] = "https://cosmos.azure.com/.default"
52
+
53
+ # AAD auth works with az login
54
+ aad_credentials = InteractiveBrowserCredential ()
55
+
56
+ # Use your credentials to authenticate your client.
57
+ aad_client = CosmosClient (HOST , aad_credentials )
58
+
59
+ # Do R/W data operations with your authorized AAD client.
60
+ db = aad_client .get_database_client (DATABASE_ID )
61
+ container = db .get_container_client (CONTAINER_ID )
62
+
63
+ # Create item
64
+ item = get_test_item (0 )
65
+ container .create_item (item )
66
+ print ("Created item:" , item ["id" ])
67
+
68
+ # Read item
69
+ read_doc = container .read_item (item = item ["id" ], partition_key = item ["pk" ])
70
+ print ("Point read:\n " + json .dumps (read_doc , indent = 2 ))
71
+
72
+
73
+ def main ():
74
+ try :
75
+ run_sample ()
76
+ except exceptions .CosmosHttpResponseError as e :
77
+ print (f"CosmosHttpResponseError: { getattr (e , 'status_code' , None )} - { e } " )
78
+ resp = getattr (e , "response" , None )
79
+ if resp is not None and getattr (resp , "headers" , None ) is not None :
80
+ try :
81
+ print ("Response headers:\n " + json .dumps (dict (resp .headers ), indent = 2 ))
82
+ except Exception :
83
+ pass
84
+ traceback .print_exc ()
85
+ raise
86
+ except Exception as ex :
87
+ print (f"Exception: { ex } " )
88
+ traceback .print_exc ()
89
+ sys .exit (1 )
90
+
91
+
92
+ if __name__ == "__main__" :
93
+ main ()
0 commit comments