Exchange authorization code for tokens.
(self, code: str)
| 114 | return code_holder["code"] |
| 115 | |
| 116 | def exchange_code(self, code: str) -> OAuthTokens | None: |
| 117 | """Exchange authorization code for tokens.""" |
| 118 | try: |
| 119 | import urllib.request |
| 120 | |
| 121 | data = urllib.parse.urlencode({ |
| 122 | "grant_type": "authorization_code", |
| 123 | "code": code, |
| 124 | "client_id": self.client_id, |
| 125 | "redirect_uri": f"http://localhost:{self.redirect_port}/callback", |
| 126 | "code_verifier": self._code_verifier, |
| 127 | }).encode("utf-8") |
| 128 | |
| 129 | req = urllib.request.Request( |
| 130 | self.token_url, |
| 131 | data=data, |
| 132 | headers={"Content-Type": "application/x-www-form-urlencoded"}, |
| 133 | ) |
| 134 | with urllib.request.urlopen(req, timeout=30) as resp: |
| 135 | body = json.loads(resp.read()) |
| 136 | |
| 137 | expires_in = body.get("expires_in", 3600) |
| 138 | return OAuthTokens( |
| 139 | access_token=body["access_token"], |
| 140 | refresh_token=body.get("refresh_token", ""), |
| 141 | token_type=body.get("token_type", "Bearer"), |
| 142 | expires_at=time.time() + expires_in, |
| 143 | scope=body.get("scope", ""), |
| 144 | ) |
| 145 | except Exception as e: |
| 146 | logger.error("OAuth token exchange failed: %s", e) |
| 147 | return None |
| 148 | |
| 149 | def refresh_tokens(self, refresh_token: str) -> OAuthTokens | None: |
| 150 | """Refresh an expired access token.""" |
nothing calls this directly
no test coverage detected