azure-functions-durable-python/azure/durable_functions/models/utils/http_utils.py at 59aba68811774fc96bcdb75cfbe5c91df2756bd1 · Azure/azure-functions-durable-python · GitHub

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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
from typing import Any, List, Union, Optional

import aiohttp


async def post_async_request(url: str,
                             data: Any = None,
                             trace_parent: str = None,
                             trace_state: str = None,
                             function_invocation_id: str = None) -> List[Union[int, Any]]:
    """Post request with the data provided to the url provided.

    Parameters
    ----------
    url: str
        url to make the post to
    data: Any
        object to post
    trace_parent: str
        traceparent header to send with the request
    trace_state: str
        tracestate header to send with the request
    function_invocation_id: str
        function invocation ID header to send for correlation

    Returns
    -------
    [int, Any]
        Tuple with the Response status code and the data returned from the request
    """
    async with aiohttp.ClientSession() as session:
        headers = {}
        if trace_parent:
            headers["traceparent"] = trace_parent
        if trace_state:
            headers["tracestate"] = trace_state
        if function_invocation_id:
            headers["X-Azure-Functions-InvocationId"] = function_invocation_id
        async with session.post(url, json=data, headers=headers) as response:
            # We disable aiohttp's input type validation
            # as the server may respond with alternative
            # data encodings. This is potentially unsafe.
            # More here: https://docs.aiohttp.org/en/stable/client_advanced.html
            data = await response.json(content_type=None)
            return [response.status, data]


async def get_async_request(url: str,
                            function_invocation_id: str = None) -> List[Any]:
    """Get the data from the url provided.

    Parameters
    ----------
    url: str
        url to get the data from
    function_invocation_id: str
        function invocation ID header to send for correlation

    Returns
    -------
    [int, Any]
        Tuple with the Response status code and the data returned from the request
    """
    async with aiohttp.ClientSession() as session:
        headers = {}
        if function_invocation_id:
            headers["X-Azure-Functions-InvocationId"] = function_invocation_id
        async with session.get(url, headers=headers) as response:
            data = await response.json(content_type=None)
            if data is None:
                data = ""
            return [response.status, data]


async def delete_async_request(url: str,
                               function_invocation_id: str = None) -> List[Union[int, Any]]:
    """Delete the data from the url provided.

    Parameters
    ----------
    url: str
        url to delete the data from
    function_invocation_id: str
        function invocation ID header to send for correlation

    Returns
    -------
    [int, Any]
        Tuple with the Response status code and the data returned from the request
    """
    async with aiohttp.ClientSession() as session:
        headers = {}
        if function_invocation_id:
            headers["X-Azure-Functions-InvocationId"] = function_invocation_id
        async with session.delete(url, headers=headers) as response:
            data = await response.json(content_type=None)
            return [response.status, data]