(
self,
api_key: typing.Union[str, None] = None,
base_url: typing.Union[str, None] = None,
timeout: typing.Union[float, None] = None,
)
| 104 | base_client: AsyncOpenPipeApi |
| 105 | |
| 106 | def __init__( |
| 107 | self, |
| 108 | api_key: typing.Union[str, None] = None, |
| 109 | base_url: typing.Union[str, None] = None, |
| 110 | timeout: typing.Union[float, None] = None, |
| 111 | ) -> None: |
| 112 | self.base_client = AsyncOpenPipeApi( |
| 113 | token="", base_url=DEFAULT_BASE_URL, timeout=timeout |
| 114 | ) |
| 115 | # set API key |
| 116 | if os.environ.get("OPENPIPE_API_KEY"): |
| 117 | self.base_client._client_wrapper._token = os.environ["OPENPIPE_API_KEY"] |
| 118 | if api_key: |
| 119 | self.base_client._client_wrapper._token = api_key |
| 120 | |
| 121 | # set base URL |
| 122 | if os.environ.get("OPENPIPE_BASE_URL"): |
| 123 | self.base_client._client_wrapper._base_url = os.environ["OPENPIPE_BASE_URL"] |
| 124 | if base_url: |
| 125 | self.base_client._client_wrapper._base_url = base_url |
| 126 | |
| 127 | @property |
| 128 | def api_key(self) -> typing.Union[str, None]: |
nothing calls this directly
no test coverage detected