-
Notifications
You must be signed in to change notification settings - Fork 2
Expand file tree
/
Copy pathasync_gateway_config.py
More file actions
108 lines (87 loc) · 3.43 KB
/
Copy pathasync_gateway_config.py
File metadata and controls
108 lines (87 loc) · 3.43 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
"""GatewayConfig resource class for asynchronous operations."""
from __future__ import annotations
from typing_extensions import Unpack, override
from ._types import BaseRequestOptions, LongRequestOptions, SDKGatewayConfigUpdateParams
from .._client import AsyncRunloop
from ..types.gateway_config_view import GatewayConfigView
class AsyncGatewayConfig:
"""Asynchronous wrapper around a gateway config resource.
Gateway configs define how to proxy API requests through the credential gateway.
They specify the target endpoint and how credentials should be applied. Use with
devboxes to securely proxy requests to external APIs without exposing API keys.
Example:
>>> runloop = AsyncRunloopSDK()
>>> gateway_config = await runloop.gateway_config.create(
... name="my-api-gateway",
... endpoint="https://api.example.com",
... auth_mechanism={"type": "bearer"},
... )
>>> info = await gateway_config.get_info()
>>> print(f"Gateway Config: {info.name}")
"""
def __init__(
self,
client: AsyncRunloop,
gateway_config_id: str,
) -> None:
"""Initialize the wrapper.
:param client: Generated AsyncRunloop client
:type client: AsyncRunloop
:param gateway_config_id: GatewayConfig ID returned by the API
:type gateway_config_id: str
"""
self._client = client
self._id = gateway_config_id
@override
def __repr__(self) -> str:
return f"<AsyncGatewayConfig id={self._id!r}>"
@property
def id(self) -> str:
"""Return the gateway config ID.
:return: Unique gateway config ID
:rtype: str
"""
return self._id
async def get_info(
self,
**options: Unpack[BaseRequestOptions],
) -> GatewayConfigView:
"""Retrieve the latest gateway config details.
Example:
>>> info = await gateway_config.get_info()
>>> print(f"Gateway Config: {info.name}, endpoint: {info.endpoint}")
:param options: Optional request configuration
:return: API response describing the gateway config
:rtype: GatewayConfigView
"""
return await self._client.gateway_configs.retrieve(
self._id,
**options,
)
async def update(self, **params: Unpack[SDKGatewayConfigUpdateParams]) -> GatewayConfigView:
"""Update the gateway config.
Example:
>>> updated = await gateway_config.update(
... name="updated-gateway-name",
... description="Updated description",
... )
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKGatewayConfigUpdateParams` for available parameters
:return: Updated gateway config view
:rtype: GatewayConfigView
"""
return await self._client.gateway_configs.update(self._id, **params)
async def delete(
self,
**options: Unpack[LongRequestOptions],
) -> GatewayConfigView:
"""Delete the gateway config. This action is irreversible.
Example:
>>> await gateway_config.delete()
:param options: Optional long-running request configuration
:return: API response acknowledging deletion
:rtype: GatewayConfigView
"""
return await self._client.gateway_configs.delete(
self._id,
**options,
)