1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
|
# ------------------------------------
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# ------------------------------------
import os
import asyncio
from azure.identity import ManagedIdentityCredential
from azure.identity.aio import ManagedIdentityCredential as AsyncManagedIdentityCredential
from azure.storage.blob import BlobServiceClient
from azure.storage.blob.aio import BlobServiceClient as AsyncBlobServiceClient
def run_sync():
credential = ManagedIdentityCredential()
credential_user_assigned = ManagedIdentityCredential(
client_id=os.environ.get("IDENTITY_USER_DEFINED_IDENTITY_CLIENT_ID")
)
client = BlobServiceClient(
account_url=f"https://{os.environ['IDENTITY_STORAGE_NAME_1']}.blob.core.windows.net",
credential=credential,
)
client2 = BlobServiceClient(
account_url=f"https://{os.environ['IDENTITY_STORAGE_NAME_2']}.blob.core.windows.net",
credential=credential_user_assigned,
)
for container in client.list_containers():
print(container["name"])
print(f"Successfully acquired token with system-assigned ManagedIdentityCredential")
for container in client2.list_containers():
print(container["name"])
print(f"Successfully acquired token with user-assigned ManagedIdentityCredential")
async def run_async():
credential = AsyncManagedIdentityCredential()
credential_user_assigned = AsyncManagedIdentityCredential(
client_id=os.environ.get("IDENTITY_USER_DEFINED_IDENTITY_CLIENT_ID")
)
client = AsyncBlobServiceClient(
account_url=f"https://{os.environ['IDENTITY_STORAGE_NAME_1']}.blob.core.windows.net",
credential=credential,
)
client2 = AsyncBlobServiceClient(
account_url=f"https://{os.environ['IDENTITY_STORAGE_NAME_2']}.blob.core.windows.net",
credential=credential_user_assigned,
)
async for container in client.list_containers():
print(container["name"])
print(f"Successfully acquired token with system-assigned async ManagedIdentityCredential")
async for container in client2.list_containers():
print(container["name"])
print(f"Successfully acquired token with user-assigned async ManagedIdentityCredential")
await client.close()
await credential.close()
await client2.close()
await credential_user_assigned.close()
if __name__ == "__main__":
run_sync()
asyncio.run(run_async())
print("Passed!")
|
