(
self,
config: ConnectionConfig,
transport: Optional[Union[BaseTransport, AsyncBaseTransport]] = None,
transport_factory: Optional[Callable[[], BaseTransport]] = None,
async_transport_factory: Optional[Callable[[], AsyncBaseTransport]] = None,
*args,
**kwargs,
)
| 103 | """ |
| 104 | |
| 105 | def __init__( |
| 106 | self, |
| 107 | config: ConnectionConfig, |
| 108 | transport: Optional[Union[BaseTransport, AsyncBaseTransport]] = None, |
| 109 | transport_factory: Optional[Callable[[], BaseTransport]] = None, |
| 110 | async_transport_factory: Optional[Callable[[], AsyncBaseTransport]] = None, |
| 111 | *args, |
| 112 | **kwargs, |
| 113 | ): |
| 114 | if transport is not None and ( |
| 115 | transport_factory is not None or async_transport_factory is not None |
| 116 | ): |
| 117 | raise ValueError("Use either transport or transport_factory, not both") |
| 118 | |
| 119 | self._transport_factory = transport_factory |
| 120 | self._async_transport_factory = async_transport_factory |
| 121 | self._thread_local = threading.local() |
| 122 | # Keyed weakly by the event loop object itself, not id(loop) — |
| 123 | # CPython reuses object ids, so a new loop could otherwise inherit |
| 124 | # a client bound to a previous, closed loop. |
| 125 | self._async_clients: weakref.WeakKeyDictionary[ |
| 126 | asyncio.AbstractEventLoop, httpx.AsyncClient |
| 127 | ] = weakref.WeakKeyDictionary() |
| 128 | self._proxy = config.proxy |
| 129 | |
| 130 | if config.api_key is None: |
| 131 | raise AuthenticationException( |
| 132 | "API key is required, please visit the API Keys tab at https://e2b.dev/dashboard?tab=keys to get your API key. " |
| 133 | "You can either set the environment variable `E2B_API_KEY` " |
| 134 | 'or you can pass it directly to the method like api_key="e2b_..."', |
| 135 | ) |
| 136 | |
| 137 | if config.api_key is not None and config.validate_api_key: |
| 138 | validate_api_key(config.api_key) |
| 139 | |
| 140 | token = config.api_key |
| 141 | auth_header_name = "X-API-KEY" |
| 142 | prefix = "" |
| 143 | |
| 144 | headers = { |
| 145 | **default_headers, |
| 146 | # Deprecated: send the access token alongside the API key when one |
| 147 | # is available, mirroring the JS SDK. Prefer `api_headers` instead. |
| 148 | # Spread before `config.headers` so a custom `Authorization` in |
| 149 | # `api_headers` wins over the deprecated access token, matching JS. |
| 150 | **( |
| 151 | {"Authorization": f"Bearer {config.access_token}"} |
| 152 | if config.access_token is not None |
| 153 | else {} |
| 154 | ), |
| 155 | **(config.headers or {}), |
| 156 | } |
| 157 | |
| 158 | # Prevent passing these parameters twice |
| 159 | more_headers: Optional[dict] = kwargs.pop("headers", None) |
| 160 | if more_headers: |
| 161 | headers.update(more_headers) |
| 162 | kwargs.pop("token", None) |
nothing calls this directly
no test coverage detected