forked from microsoftgraph/msgraph-sdk-python-core
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathgraph_client_factory.py
More file actions
141 lines (124 loc) · 6.07 KB
/
graph_client_factory.py
File metadata and controls
141 lines (124 loc) · 6.07 KB
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
# ------------------------------------
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# ------------------------------------
from __future__ import annotations
from typing import Optional
import httpx
from kiota_abstractions.request_option import RequestOption
from kiota_http.kiota_client_factory import KiotaClientFactory
from kiota_http.middleware.middleware import BaseMiddleware
from ._enums import APIVersion, NationalClouds
from .middleware import AsyncGraphTransport, GraphTelemetryHandler
from .middleware.options import GraphTelemetryHandlerOption
class GraphClientFactory():
"""Constructs httpx.AsyncClient instances configured with either custom or default
pipeline of graph specific middleware.
"""
@staticmethod
def create_with_default_middleware(
api_version: APIVersion = APIVersion.v1,
client: Optional[httpx.AsyncClient] = None,
host: NationalClouds = NationalClouds.Global,
options: Optional[dict[str, RequestOption]] = None
) -> httpx.AsyncClient:
"""Constructs native HTTP AsyncClient(httpx.AsyncClient) instances configured with
a custom transport loaded with a default pipeline of middleware.
Args:
api_version (APIVersion): The Graph API version to be used.
Defaults to APIVersion.v1.
This is only used if the client parameter is None.
client (Optional[httpx.AsyncClient]]): The httpx.AsyncClient instance to be used.
Defaults to None.
When None, a client will be created with base url set to https://{host}/{api_version}.
host (NationalClouds): The national clound endpoint to be used.
Defaults to NationalClouds.Global.
This is only used if the client parameter is None.
options (Optional[dict[str, RequestOption]]): The request options to use when
instantiating default middleware. Defaults to dict[str, RequestOption]=None.
Returns:
httpx.AsyncClient: An instance of the AsyncClient object
"""
if client is None:
client = KiotaClientFactory.get_default_client()
client.base_url = GraphClientFactory._get_base_url(host, api_version)
middleware = KiotaClientFactory.get_default_middleware(options)
telemetry_handler = GraphClientFactory._get_telemetry_handler(options)
middleware.append(telemetry_handler)
return GraphClientFactory._load_middleware_to_client(client, middleware)
@staticmethod
def create_with_custom_middleware(
middleware: Optional[list[BaseMiddleware]],
api_version: APIVersion = APIVersion.v1,
client: Optional[httpx.AsyncClient] = None,
host: NationalClouds = NationalClouds.Global,
) -> httpx.AsyncClient:
"""Applies a custom middleware chain to the HTTP Client
Args:
middleware(Optional[list[BaseMiddleware]]): Custom middleware list that will be used to
create a middleware pipeline. The middleware should be arranged in the order in which
they will modify the request.
Defaults to None,
api_version (APIVersion): The Graph API version to be used.
Defaults to APIVersion.v1.
This is only used if the client parameter is None.
client (Optional[httpx.AsyncClient]): The httpx.AsyncClient instance to be used.
Defaults to None.
When None, a client will be created with base url set to https://{host}/{api_version}.
host (NationalClouds): The national cloud endpoint to be used.
Defaults to NationalClouds.Global.
This is only used if the client parameter is None.
"""
if client is None:
client = KiotaClientFactory.get_default_client()
client.base_url = GraphClientFactory._get_base_url(host, api_version)
return GraphClientFactory._load_middleware_to_client(client, middleware)
@staticmethod
def _get_base_url(host: str, api_version: APIVersion) -> str:
"""Helper method to set the complete base url"""
base_url = f'{host}/{api_version}'
return base_url
@staticmethod
def _get_telemetry_handler(
options: Optional[dict[str, RequestOption]]
) -> GraphTelemetryHandler:
"""Helper method to get the graph telemetry handler instantiated with appropriate
options"""
if options:
graph_telemetry_options: GraphTelemetryHandlerOption = options.get(
GraphTelemetryHandlerOption().get_key()
) # type: ignore # Unable to down cast type
if graph_telemetry_options:
return GraphTelemetryHandler(options=graph_telemetry_options)
return GraphTelemetryHandler()
@staticmethod
def _load_middleware_to_client(
client: httpx.AsyncClient, middleware: Optional[list[BaseMiddleware]]
) -> httpx.AsyncClient:
current_transport = client._transport
client._transport = GraphClientFactory._replace_transport_with_custom_graph_transport(
current_transport, middleware
)
if client._mounts:
mounts: dict = {}
for pattern, transport in client._mounts.items():
if transport is None:
mounts[pattern] = None
else:
mounts[pattern
] = GraphClientFactory._replace_transport_with_custom_graph_transport(
transport, middleware
)
client._mounts = dict(sorted(mounts.items()))
return client
@staticmethod
def _replace_transport_with_custom_graph_transport(
current_transport: httpx.AsyncBaseTransport, middleware: Optional[list[BaseMiddleware]]
) -> AsyncGraphTransport:
middleware_pipeline = KiotaClientFactory.create_middleware_pipeline(
middleware, current_transport
)
new_transport = AsyncGraphTransport(
transport=current_transport, pipeline=middleware_pipeline
)
return new_transport