目前遇到的困難

Functions 可以做很多事情...
但遇到如果要操作azure 某些 resource...
勢必會用到azure cli

但要怎怎麼在python function 直接使用azure cli呢?

Solution

其實所有的方法, 底層都是使用一樣的方式
基本上都還是call restful api

1. 直接找 restful api 文件 改用requests module

• https://docs.microsoft.com/en-us/rest/api/azure/
• https://docs.microsoft.com/en-us/rest/api/azure/devops/?view=azure-devops-rest-6.1

def add_agent_user_capability(self, agent, new_capabilities):
        pool_id = self._get_pool_id_by_name(self.pool_name)
        agent_id = agent.id
        status_info = defaultdict(list)
        url = f"https://dev.azure.com/infinite-wars/_apis/distributedtask/pools/{pool_id}/agents/{agent_id}/usercapabilities?api-version=5.1"

        for key, value in new_capabilities.items():
            status_info[key] = value
        logging.debug(f"new status_info: {status_info}")
        rtn = requests.put(url, json=status_info, auth=HTTPBasicAuth('', self.credential.personal_access_token))
        assert rtn.status_code == 200

2. 使用對應的python module, 例如 azure-devops

import json

from azure.devops.connection import Connection
from azure.devops.v6_0.release.models import ReleaseStartMetadata
from azure.devops.v6_0.release.models import ConfigurationVariableValue
from msrest.authentication import BasicAuthentication


class ReleasePipelineNotFound(Exception):
    pass


class RunReleasePipeline:
    def __init__(self, credential):
        credentials = BasicAuthentication('', credential.personal_access_token)
        connection = Connection(base_url=credential.organization_url, creds=credentials)
        self.release_client = connection.clients_v6_0.get_release_client()
        self.credential = credential

3. 假設真的必須只有azure-command-line能使用的話, 可以在function app 使用 azure.cli.core module

import logging

import azure.functions as func
from azure.cli.core import get_default_cli


def az_cli(args):
    cli = get_default_cli()
    cli.invoke(args)

    if cli.result.result:
        return cli.result.result

    elif cli.result.error:
        raise cli.result.error

    return True

然後可以在function中直接使用

def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')

    name = req.params.get('name')
    if not name:
        try:
            req_body = req.get_json()
        except ValueError:
            pass
        else:
            name = req_body.get('name')

    if name:
        return func.HttpResponse(f"Hello, {name}. This HTTP triggered function executed successfully.")
    else:

        command = ['lab', 'get', '-g', 'rg-testing-env-lab', '--name', 'dtl-aladdin-test']
        test = az_cli(command)
        logging.info(f"test return {test}")

        logging.info("123456")

        return func.HttpResponse(
             "This HTTP triggered function executed successfully. Pass a name in the query string or in the request body for a personalized response.",
             status_code=200
        )

就可以使用囉

Notes

但假設使用的api有再利用subprocess去call command 或是使用tool的話 就會失敗! 這點必須注意~~

在Function執行下面command

command = ['artifacts', 'universal', 'download', '--organization', 'https://dev.azure.com/infinite-wars/',
                   '--feed', 'infinite-wars', '--name', 'data-l2-testing-case-v1.0.1776', '--version', '1.0.0',
                   '--path', '.']
        a = az_cli(command)
        logging.info(f"a return {a}")

會得到

[2021-03-27T09:33:27.378Z] extension name: azure-devops
[2021-03-27T09:33:27.380Z] extension version: 0.18.0
[2021-03-27T09:33:27.435Z] signal only works in main thread
[2021-03-27T09:33:27.436Z] exit code: 1
[2021-03-27T09:33:27.441Z] Executed 'Functions.func_testing' (Failed, Id=d52b77a2-1f61-487f-8f3b-c7948b1b1605, Duration=1106ms)
[2021-03-27T09:33:27.443Z] System.Private.CoreLib: Exception while executing function: 
Functions.func_testing. System.Private.CoreLib: Result: Failure
Exception: CLIError: signal only works in main thread
Stack:   File "C:\ProgramData\chocolatey\lib\azure-functions-core-tools\tools\workers\python\3.8/WINDOWS/X64\azure_functions_worker\dispatcher.py", line 355, in _handle__invocation_request
    call_result = await self._loop.run_in_executor(
  File "C:\Python38\lib\concurrent\futures\thread.py", line 57, in run
    result = self.fn(*self.args, **self.kwargs)
  File "C:\ProgramData\chocolatey\lib\azure-functions-core-tools\tools\workers\python\3.8/WINDOWS/X64\azure_functions_worker\dispatcher.py", line 542, in __run_sync_func     
    return func(**params)
  File "E:\Coding\practice\func_app\func_testing\__init__.py", line 39, in main        
    a = az_cli(command)
  File "E:\Coding\practice\func_app\func_testing\__init__.py", line 15, in az_cli      
    raise cli.result.error
  File "E:\Coding\practice\func_app\.venv\lib\site-packages\knack\cli.py", line 233, in invoke
    cmd_result = self.invocation.execute(args)
  File "E:\Coding\practice\func_app\.venv\lib\site-packages\azure\cli\core\commands\__init__.py", line 660, in execute
    raise ex
  File "E:\Coding\practice\func_app\.venv\lib\site-packages\azure\cli\core\commands\__init__.py", line 723, in _run_jobs_serially
    results.append(self._run_job(expanded_arg, cmd_copy))
  File "E:\Coding\practice\func_app\.venv\lib\site-packages\azure\cli\core\commands\__init__.py", line 715, in _run_job
    return cmd_copy.exception_handler(ex)
  File "C:\Users\Jerry He\.azure\cliextensions\azure-devops\azext_devops\dev\common\exception_handler.py", line 26, in azure_devops_exception_handler
    raise CLIError(ex)

https://github.com/Azure/azure-devops-cli-extension/blob/master/azure-devops/azext_devops/dev/artifacts/universal.py

仔細去看完source code就會發現 他其實是去call另外一條thread去做事 但目前function好像不support這樣的事情

def run(self, command_args, env, initial_progress_text, stderr_handler):
        with humanfriendly.Spinner(  # pylint: disable=no-member
                label=initial_progress_text, total=100, stream=sys.stderr) as self._spinner:
            self._spinner.step()
            # Start the process, process stderr for progress reporting, check the process result
            self.start(command_args, env)
            try:
                for bline in iter(self._proc.stderr.readline, b''):
                    line = bline.decode('utf-8', 'ignore').strip()
                    stderr_handler(line, self._update_progress)
                return self.wait()
            except IOError as ex:
                if not self._terminating:
                    raise ex

Reference

• https://stackoverflow.com/questions/57405394/azure-function-trigger-python-script-containing-azure-cli-commands/57458509#57458509
• https://github.com/jerryeml/func_app