代码之家 › 专栏 › 技术社区 › SodaCris

无法使用Azure Python SDK获取Azure订阅

azure-python-sdk azure-management-api azure python

0

SodaCris · 技术社区 · 2 年前

我可以获取Azure下标列表 using REST API .

但是,当我切换到 Azure Python SDK ,似乎有一些问题。

这是迄今为止的代码:

from azure.identity.aio import ClientSecretCredential
from azure.mgmt.resource import SubscriptionClient
import json

data = json.load(open("parameters.json"))

credential = ClientSecretCredential(
    tenant_id=data["tenant"],
    client_id=data["client_id"],
    client_secret=data["client_secret"],
)

subs = SubscriptionClient(credential=credential)
l = list(subs.subscriptions.list())
print(l)

我使用一个附加 list 在pnultimate行中,因为 subs.subscriptions.list() 返回一个迭代器。尽管如此,代码似乎相当简单。

但是,此代码会出现以下错误:

Traceback (most recent call last):
  File "c:\Users\azureuser\Documents\GitHub\vmss-scripts\vm_create.py", line 14, in <module>
    l = list(subs.subscriptions.list())
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\azureuser\scoop\apps\python\current\Lib\site-packages\azure\core\paging.py", line 123, in __next__
    return next(self._page_iterator)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\azureuser\scoop\apps\python\current\Lib\site-packages\azure\core\paging.py", line 75, in __next__
    self._response = self._get_next(self.continuation_token)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\azureuser\scoop\apps\python\current\Lib\site-packages\azure\mgmt\resource\subscriptions\v2022_12_01\operations\_operations.py", line 526, in get_next
    pipeline_response: PipelineResponse = self._client._pipeline.run(  # pylint: disable=protected-access
                                          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\azureuser\scoop\apps\python\current\Lib\site-packages\azure\core\pipeline\_base.py", line 230, in run
    return first_node.send(pipeline_request)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\azureuser\scoop\apps\python\current\Lib\site-packages\azure\core\pipeline\_base.py", line 86, in send
    response = self.next.send(request)
               ^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\azureuser\scoop\apps\python\current\Lib\site-packages\azure\core\pipeline\_base.py", line 86, in send
    response = self.next.send(request)
               ^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\azureuser\scoop\apps\python\current\Lib\site-packages\azure\core\pipeline\_base.py", line 86, in send
    response = self.next.send(request)
               ^^^^^^^^^^^^^^^^^^^^^^^
  [Previous line repeated 2 more times]
  File "C:\Users\azureuser\scoop\apps\python\current\Lib\site-packages\azure\mgmt\core\policies\_base.py", line 46, in send
    response = self.next.send(request)
               ^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\azureuser\scoop\apps\python\current\Lib\site-packages\azure\core\pipeline\policies\_redirect.py", line 197, in send
    response = self.next.send(request)
               ^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\azureuser\scoop\apps\python\current\Lib\site-packages\azure\core\pipeline\policies\_retry.py", line 531, in send
    response = self.next.send(request)
               ^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\azureuser\scoop\apps\python\current\Lib\site-packages\azure\core\pipeline\policies\_authentication.py", line 124, in send
    self.on_request(request)
  File "C:\Users\azureuser\scoop\apps\python\current\Lib\site-packages\azure\core\pipeline\policies\_authentication.py", line 100, in on_request
    self._update_headers(request.http_request.headers, self._token.token)
                                                       ^^^^^^^^^^^^^^^^^
AttributeError: 'coroutine' object has no attribute 'token'
sys:1: RuntimeWarning: coroutine 'GetTokenMixin.get_token' was never awaited

我不知道遗漏了什么,这个错误也没有提供太多信息。

似乎与有关 token 。我可以使用以下方法获取令牌:

async def print_token():
    token = await credential.get_token("https://management.azure.com/.default")
    print(token.token)
    await credential.close()

asyncio.run(print_token())

但这需要 asyncio 运行,这与我的代码不兼容。我应该把代币放在哪里?

我已经调查了源代码,如果 Azure CLI 。它似乎使用相同的方法获取订阅: https://github.com/Azure/azure-cli/blob/f369cead2604e37480611b0cc269fee615956ea2/src/azure-cli-core/azure/cli/core/_profile.py#L835

客户端是从下面的函数中获取的,类型为

https://github.com/Azure/azure-cli/blob/f369cead2604e37480611b0cc269fee615956ea2/src/azure-cli-core/azure/cli/core/profiles/_shared.py#L60

本质上是 SubscriptionClient .

2 回复 | 直到 2 年前

1

Sridevi 2 年前

起初,我也得到了 相同的错误 当我在我的环境中这样运行您的代码时:

from azure.identity.aio import ClientSecretCredential
from azure.mgmt.resource import SubscriptionClient
import json

data = json.load(open("parameters.json"))

credential = ClientSecretCredential(
    tenant_id=data["tenant"],
    client_id=data["client_id"],
    client_secret=data["client_secret"],
)

subs = SubscriptionClient(credential=credential)
l = list(subs.subscriptions.list())
print(l)

回答

使用Azure Python SDK列出订阅,不使用 asyncio 模块,使用以下修改后的代码:

from azure.identity import ClientSecretCredential
from azure.mgmt.resource import SubscriptionClient
import json

with open("parameters.json") as f:
    data = json.load(f)

credential = ClientSecretCredential(
    tenant_id=data["tenant"],
    client_id=data["client_id"],
    client_secret=data["client_secret"],
)

subs_client = SubscriptionClient(credential)

subscriptions = subs_client.subscriptions.list()

for subscription in subscriptions:
    #print(f"Subscription ID: {subscription.subscription_id}, Display Name: {subscription.display_name}")
    print(json.dumps(subscription.as_dict(), indent=2))

回答

2

0

SodaCris 2 年前

此外,这是异步版本的代码。

import asyncio
from azure.identity.aio import ClientSecretCredential
from azure.mgmt.resource.subscriptions.aio import SubscriptionClient
import json

data = None
with open("parameters.json") as f:
    data = json.load(f)


async def app():
    credential = ClientSecretCredential(
        tenant_id=data["tenant"],
        client_id=data["client_id"],
        client_secret=data["client_secret"],
    )

    subs_client = SubscriptionClient(credential)

    subscriptions = subs_client.subscriptions.list()

    async for subscription in subscriptions:
        # print(f"Subscription ID: {subscription.subscription_id}, Display Name: {subscription.display_name}")
        print(json.dumps(subscription.as_dict(), indent=2))

    await subs_client.close()
    await credential.close()


asyncio.run(app())