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
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
|
# ---------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# ---------------------------------------------------------
from functools import wraps
from typing import Any, Callable, Optional
from typing_extensions import Concatenate, ParamSpec, Self
from azure.core.configuration import Configuration
from azure.core.pipeline import Pipeline
from azure.core.pipeline.policies import (
CustomHookPolicy,
HeadersPolicy,
HttpLoggingPolicy,
NetworkTraceLoggingPolicy,
ProxyPolicy,
RedirectPolicy,
RetryPolicy,
UserAgentPolicy,
)
from azure.core.pipeline.transport import ( # pylint: disable=non-abstract-transport-import,no-name-in-module
HttpTransport,
RequestsTransport,
)
from azure.core.rest import HttpRequest, HttpResponse
def _request_function(f: Callable[["HttpPipeline"], None]):
"""A decorator that provides the implementation for the request-style convenience functions of the HttpPipeline
class.
:param Callable[[], None] f: A function whose name will be used as the http
request method
:return: An HTTP request function
:rtype: Callable
"""
# This is a hack to provide richer typing for the decorated function
# The outer _ function allows us to pattern match on the parameter types
# of HttpRequest and forward those types to the decorated function
P = ParamSpec("P")
def _(_: Callable[Concatenate[str, P], Any] = HttpRequest):
@wraps(f)
# pylint: disable-next=docstring-missing-param
def decorated(self: "HttpPipeline", *args: P.args, **kwargs: P.kwargs) -> HttpResponse:
"""A function that sends an HTTP request and returns the response.
Accepts the same parameters as azure.core.rest.HttpRequest, except for the method.
All other kwargs are forwarded to azure.core.Pipeline.run
:return: The request response
:rtype: HttpResponse
"""
request = HttpRequest(
f.__name__.upper(),
*args,
params=kwargs.pop("params", None),
headers=kwargs.pop("headers", None),
json=kwargs.pop("json", None),
content=kwargs.pop("content", None),
data=kwargs.pop("data", None),
files=kwargs.pop("files", None),
)
return self.run(request, **kwargs).http_response
return decorated
return _()
class HttpPipeline(Pipeline):
"""A *very* thin wrapper over azure.core.pipeline.Pipeline that facilitates sending miscellaneous http requests by
adding:
* A requests-style api for sending http requests
* Facilities for populating policies for the client, include defaults,
and re-using policies from an existing client.
"""
def __init__(
self,
*,
transport: Optional[HttpTransport] = None,
config: Optional[Configuration] = None,
user_agent_policy: Optional[UserAgentPolicy] = None,
headers_policy: Optional[HeadersPolicy] = None,
proxy_policy: Optional[ProxyPolicy] = None,
logging_policy: Optional[NetworkTraceLoggingPolicy] = None,
http_logging_policy: Optional[HttpLoggingPolicy] = None,
retry_policy: Optional[RetryPolicy] = None,
custom_hook_policy: Optional[CustomHookPolicy] = None,
redirect_policy: Optional[RedirectPolicy] = None,
**kwargs
):
"""
:param HttpTransport transport: Http Transport used for requests, defaults to RequestsTransport
:param Configuration config:
:param UserAgentPolicy user_agent_policy:
:param HeadersPolicy headers_policy:
:param ProxyPolicy proxy_policy:
:param NetworkTraceLoggingPolicy logging_policy:
:param HttpLoggingPolicy http_logging_policy:
:param RetryPolicy retry_policy:
:param CustomHookPolicy custom_hook_policy:
:param RedirectPolicy redirect_policy:
"""
config = config or Configuration()
config.headers_policy = headers_policy or config.headers_policy or HeadersPolicy(**kwargs)
config.proxy_policy = proxy_policy or config.proxy_policy or ProxyPolicy(**kwargs)
config.redirect_policy = redirect_policy or config.redirect_policy or RedirectPolicy(**kwargs)
config.retry_policy = retry_policy or config.retry_policy or RetryPolicy(**kwargs)
config.custom_hook_policy = custom_hook_policy or config.custom_hook_policy or CustomHookPolicy(**kwargs)
config.logging_policy = logging_policy or config.logging_policy or NetworkTraceLoggingPolicy(**kwargs)
config.http_logging_policy = http_logging_policy or config.http_logging_policy or HttpLoggingPolicy(**kwargs)
config.user_agent_policy = user_agent_policy or config.user_agent_policy or UserAgentPolicy(**kwargs)
config.polling_interval = kwargs.get("polling_interval", 30)
super().__init__(
# RequestsTransport normally should not be imported outside of azure.core, since transports
# are meant to be user configurable.
# RequestsTransport is only used in this file as the default transport when not user specified.
transport=transport or RequestsTransport(**kwargs),
policies=[
config.headers_policy,
config.user_agent_policy,
config.proxy_policy,
config.redirect_policy,
config.retry_policy,
config.authentication_policy,
config.custom_hook_policy,
config.logging_policy,
],
)
self._config = config
def with_policies(self, **kwargs) -> Self:
"""A named constructor which facilitates creating a new pipeline using an existing one as a base.
Accepts the same parameters as __init__
:return: new Pipeline object with combined config of current object
and specified overrides
:rtype: Self
"""
cls = self.__class__
return cls(config=self._config, transport=kwargs.pop("transport", self._transport), **kwargs)
@_request_function
def delete(self) -> None:
"""Sends a DELETE request."""
@_request_function
def get(self) -> None:
"""Sends a GET request."""
@_request_function
def head(self) -> None:
"""Sends a HEAD request."""
@_request_function
def options(self) -> None:
"""Sends a OPTIONS request."""
@_request_function
def patch(self) -> None:
"""Sends a PATCH request."""
@_request_function
def post(self) -> None:
"""Sends a POST request."""
@_request_function
def put(self) -> None:
"""Sends a PUT request."""
|
