estel/dify
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

chore(api/libs): Apply ruff format. (#7301)

Browse Source
This commit is contained in:
-LAN-
2024-08-15 17:53:12 +08:00
committed by GitHub
parent d07b2b9915
commit 9414143b5f
15 changed files with 270 additions and 320 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
api/libs/oauth.py
View File

@@ -35,31 +35,31 @@ class OAuth:
class GitHubOAuth(OAuth):
_AUTH_URL = 'https://github.com/login/oauth/authorize'
_TOKEN_URL = 'https://github.com/login/oauth/access_token'
_USER_INFO_URL = 'https://api.github.com/user'
_EMAIL_INFO_URL = 'https://api.github.com/user/emails'
_AUTH_URL = "https://github.com/login/oauth/authorize"
_TOKEN_URL = "https://github.com/login/oauth/access_token"
_USER_INFO_URL = "https://api.github.com/user"
_EMAIL_INFO_URL = "https://api.github.com/user/emails"
def get_authorization_url(self):
params = {
'client_id': self.client_id,
'redirect_uri': self.redirect_uri,
'scope': 'user:email' # Request only basic user information
"client_id": self.client_id,
"redirect_uri": self.redirect_uri,
"scope": "user:email", # Request only basic user information
}
return f"{self._AUTH_URL}?{urllib.parse.urlencode(params)}"
def get_access_token(self, code: str):
data = {
'client_id': self.client_id,
'client_secret': self.client_secret,
'code': code,
'redirect_uri': self.redirect_uri
"client_id": self.client_id,
"client_secret": self.client_secret,
"code": code,
"redirect_uri": self.redirect_uri,
}
headers = {'Accept': 'application/json'}
headers = {"Accept": "application/json"}
response = requests.post(self._TOKEN_URL, data=data, headers=headers)
response_json = response.json()
access_token = response_json.get('access_token')
access_token = response_json.get("access_token")
if not access_token:
raise ValueError(f"Error in GitHub OAuth: {response_json}")
@@ -67,55 +67,51 @@ class GitHubOAuth(OAuth):
return access_token
def get_raw_user_info(self, token: str):
headers = {'Authorization': f"token {token}"}
headers = {"Authorization": f"token {token}"}
response = requests.get(self._USER_INFO_URL, headers=headers)
response.raise_for_status()
user_info = response.json()
email_response = requests.get(self._EMAIL_INFO_URL, headers=headers)
email_info = email_response.json()
primary_email = next((email for email in email_info if email['primary'] == True), None)
primary_email = next((email for email in email_info if email["primary"] == True), None)
return {**user_info, 'email': primary_email['email']}
return {**user_info, "email": primary_email["email"]}
def _transform_user_info(self, raw_info: dict) -> OAuthUserInfo:
email = raw_info.get('email')
email = raw_info.get("email")
if not email:
email = f"{raw_info['id']}+{raw_info['login']}@users.noreply.github.com"
return OAuthUserInfo(
id=str(raw_info['id']),
name=raw_info['name'],
email=email
)
return OAuthUserInfo(id=str(raw_info["id"]), name=raw_info["name"], email=email)
class GoogleOAuth(OAuth):
_AUTH_URL = 'https://accounts.google.com/o/oauth2/v2/auth'
_TOKEN_URL = 'https://oauth2.googleapis.com/token'
_USER_INFO_URL = 'https://www.googleapis.com/oauth2/v3/userinfo'
_AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth"
_TOKEN_URL = "https://oauth2.googleapis.com/token"
_USER_INFO_URL = "https://www.googleapis.com/oauth2/v3/userinfo"
def get_authorization_url(self):
params = {
'client_id': self.client_id,
'response_type': 'code',
'redirect_uri': self.redirect_uri,
'scope': 'openid email'
"client_id": self.client_id,
"response_type": "code",
"redirect_uri": self.redirect_uri,
"scope": "openid email",
}
return f"{self._AUTH_URL}?{urllib.parse.urlencode(params)}"
def get_access_token(self, code: str):
data = {
'client_id': self.client_id,
'client_secret': self.client_secret,
'code': code,
'grant_type': 'authorization_code',
'redirect_uri': self.redirect_uri
"client_id": self.client_id,
"client_secret": self.client_secret,
"code": code,
"grant_type": "authorization_code",
"redirect_uri": self.redirect_uri,
}
headers = {'Accept': 'application/json'}
headers = {"Accept": "application/json"}
response = requests.post(self._TOKEN_URL, data=data, headers=headers)
response_json = response.json()
access_token = response_json.get('access_token')
access_token = response_json.get("access_token")
if not access_token:
raise ValueError(f"Error in Google OAuth: {response_json}")
@@ -123,16 +119,10 @@ class GoogleOAuth(OAuth):
return access_token
def get_raw_user_info(self, token: str):
headers = {'Authorization': f"Bearer {token}"}
headers = {"Authorization": f"Bearer {token}"}
response = requests.get(self._USER_INFO_URL, headers=headers)
response.raise_for_status()
return response.json()
def _transform_user_info(self, raw_info: dict) -> OAuthUserInfo:
return OAuthUserInfo(
id=str(raw_info['sub']),
name=None,
email=raw_info['email']
)
return OAuthUserInfo(id=str(raw_info["sub"]), name=None, email=raw_info["email"])