| 101 | |
| 102 | |
| 103 | class AsyncOpenPipe: |
| 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]: |
| 129 | """Property getter for api_key.""" |
| 130 | return self.base_client._client_wrapper._token |
| 131 | |
| 132 | @api_key.setter |
| 133 | def api_key(self, value: typing.Union[str, None]) -> None: |
| 134 | """Property setter for api_key.""" |
| 135 | self._api_key = value |
| 136 | if value is not None: |
| 137 | self.base_client._client_wrapper._token = value |
| 138 | |
| 139 | @property |
| 140 | def base_url(self) -> typing.Union[str, None]: |
| 141 | """Property getter for base_url.""" |
| 142 | return self.base_client._client_wrapper._base_url |
| 143 | |
| 144 | @base_url.setter |
| 145 | def base_url(self, value: typing.Union[str, None]) -> None: |
| 146 | """Property setter for base_url.""" |
| 147 | if value is not None: |
| 148 | self.base_client._client_wrapper._base_url = value |
| 149 | |
| 150 | async def report( |
| 151 | self, |
| 152 | *, |
| 153 | requested_at: typing.Optional[float] = OMIT, |
| 154 | received_at: typing.Optional[float] = OMIT, |
| 155 | req_payload: typing.Optional[typing.Any] = OMIT, |
| 156 | resp_payload: typing.Optional[typing.Any] = OMIT, |
| 157 | status_code: typing.Optional[float] = OMIT, |
| 158 | error_message: typing.Optional[str] = OMIT, |
| 159 | tags: typing.Optional[typing.Dict[str, ReportRequestTagsValue]] = {}, |
| 160 | ) -> ReportResponse: |
no outgoing calls