python - How to implement REST API call to Azure Functions? - Stack Overflow

admin管理员组

文章数量:1336660

I am trying to create an Azure Function that runs every time a list of users is updated on a third-party REST API. This is how my code looks on its own, and works as expected (returns the result I'm looking for):

import base64
import requests
from urllib.parse import urlencode
import os


username = os.environ["username"]
password = os.environ["password"]

encoded_username = urlencode({"username": username}).split('=')[1]
encoded_password = urlencode({"password": password}).split('=')[1]

# Combine username and password into a single string
url_cred = f"{encoded_username}:{encoded_password}"

# Encode the combined string in Base64
encoded_b64 = base64.b64encode(url_cred.encode('utf-8')).decode('utf-8')

# Set up headers for the POST request
headers = {
    "authorization": f"Basic {encoded_b64}"
}

# Set up body for the POST request
body = {
    "grant_type": "client_credentials"
}

# Obtain access token with GET request
response = requests.post(";, headers=headers, data=body)
token = response.json().get('access_token')

# Set up headers for the GET request with the access token
auth_header = {
    "authorization": f"Bearer {token}"
}

# Make the GET request to retrieve users
response = requests.get(";, headers=auth_header)
# ! This is the GET request I want to use to trigger the function !
results = response.json()

username_results = []
for data in results:
    for elem in results["usersResult"]:
        username_results.append(elem['userName'])


for name in username_results:
    print(name)

Here is my attempt at integrating the code into an Azure Function:

import azure.functions as func
import logging
import json
import requests
import base64
from urllib.parse import urlencode
import os


app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)

username = os.environ["username"]
password = os.environ["password"]


@app.route(route="api_call")
def api_call(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')

    # URL encode username and password
    encoded_username = urlencode({"username": username}).split("=")[1]
    encoded_password = urlencode({"password": password}).split("=")[1]

    url_cred = f"{encoded_username}:{encoded_password}"

    # Encode in Base64
    encoded_64 = base64.b64encode(url_cred.encode("utf-8")).decode("utf-8")

    # Headers for POST request
    headers = {
        "authorization": f"Basic {encoded_64}"
    }

    # Body for POST request
    body = {
        "grant_type": "client_credentials"
    }

    # Obtain access token with GET request
    response = requests.post(";, headers=headers, data=body)
    token = response.json().get("access_token")

    # Headers for GET request with access token
    auth_header = {
        "authorization": f"Bearer {token}"
    }

    # Make GET request to retrieve users
    res = requests.get(";, headers=auth_header)

   # ! The GET request I want to use to trigger the function !
    results = req.get_json(res)


    # Filter results and return only usernames
    username_results = []
    for data in results:
        for elem in results["usersResult"]:
            username_results.append(elem["userName"])

    for name in username_results:
        return func.HttpResponse(json.dumps(name))
  

With this code, I get the error System.Private.CoreLib: Result: Failure Exception: TypeError: HttpRequest.get_json() takes 1 positional argument but 2 were given

How do I connect the HTTP trigger correctly?

I have read the documentations on Azure Function HTTP triggers and bindings but can't find any good examples to help.

I am trying to create an Azure Function that runs every time a list of users is updated on a third-party REST API. This is how my code looks on its own, and works as expected (returns the result I'm looking for):

import base64
import requests
from urllib.parse import urlencode
import os


username = os.environ["username"]
password = os.environ["password"]

encoded_username = urlencode({"username": username}).split('=')[1]
encoded_password = urlencode({"password": password}).split('=')[1]

# Combine username and password into a single string
url_cred = f"{encoded_username}:{encoded_password}"

# Encode the combined string in Base64
encoded_b64 = base64.b64encode(url_cred.encode('utf-8')).decode('utf-8')

# Set up headers for the POST request
headers = {
    "authorization": f"Basic {encoded_b64}"
}

# Set up body for the POST request
body = {
    "grant_type": "client_credentials"
}

# Obtain access token with GET request
response = requests.post("https://examplewebsite/oauth2/token", headers=headers, data=body)
token = response.json().get('access_token')

# Set up headers for the GET request with the access token
auth_header = {
    "authorization": f"Bearer {token}"
}

# Make the GET request to retrieve users
response = requests.get("https://examplewebsite", headers=auth_header)
# ! This is the GET request I want to use to trigger the function !
results = response.json()

username_results = []
for data in results:
    for elem in results["usersResult"]:
        username_results.append(elem['userName'])


for name in username_results:
    print(name)

Here is my attempt at integrating the code into an Azure Function:

import azure.functions as func
import logging
import json
import requests
import base64
from urllib.parse import urlencode
import os


app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)

username = os.environ["username"]
password = os.environ["password"]


@app.route(route="api_call")
def api_call(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')

    # URL encode username and password
    encoded_username = urlencode({"username": username}).split("=")[1]
    encoded_password = urlencode({"password": password}).split("=")[1]

    url_cred = f"{encoded_username}:{encoded_password}"

    # Encode in Base64
    encoded_64 = base64.b64encode(url_cred.encode("utf-8")).decode("utf-8")

    # Headers for POST request
    headers = {
        "authorization": f"Basic {encoded_64}"
    }

    # Body for POST request
    body = {
        "grant_type": "client_credentials"
    }

    # Obtain access token with GET request
    response = requests.post("https://examplewebsite/oauth2/token", headers=headers, data=body)
    token = response.json().get("access_token")

    # Headers for GET request with access token
    auth_header = {
        "authorization": f"Bearer {token}"
    }

    # Make GET request to retrieve users
    res = requests.get("https://examplewebsite", headers=auth_header)

   # ! The GET request I want to use to trigger the function !
    results = req.get_json(res)


    # Filter results and return only usernames
    username_results = []
    for data in results:
        for elem in results["usersResult"]:
            username_results.append(elem["userName"])

    for name in username_results:
        return func.HttpResponse(json.dumps(name))
  

With this code, I get the error System.Private.CoreLib: Result: Failure Exception: TypeError: HttpRequest.get_json() takes 1 positional argument but 2 were given

How do I connect the HTTP trigger correctly?

I have read the documentations on Azure Function HTTP triggers and bindings but can't find any good examples to help.

• python
• azure-functions
• azure-http-trigger

Share Improve this question Follow asked Nov 19, 2024 at 15:33 tsontson 1 1

• Did you deploy it to azure? – RithwikBojja Commented Nov 20, 2024 at 3:30

Add a comment  | 

1 Answer 1

Sorted by: Reset to default 0

ystem.Private.CoreLib: Result: Failure Exception: TypeError: HttpRequest.get_json() takes 1 positional argument but 2 were given

The get_json() just parses the request body of http function.

So you need to change this results = req.get_json(res) to: results = res.json() in code then error goes off.

get_json() (get the body of request) example:

import azure.functions as func
import logging as rlg

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@app.route(route="http_trigger")
def http_trigger(req: func.HttpRequest) -> func.HttpResponse:
    rlg.info('Python HTTP trigger function processed a request.')
    rith = req.get_json()
    rlg.info(f"Hello Rithwik, the json is : {rith}")
    return func.HttpResponse(f"{rith}",status_code=200)

So use get_json() only when you want to get it function app's http request body and not for other http requests.

本文标签： pythonHow to implement REST API call to Azure FunctionsStack Overflow