作者:香柏林茶业 | 来源:互联网 | 2023-07-29 17:05
importosimporttimeimportrequestsasrest_clientfromurllib.parseimporturljoinfromfastapiimpor
import os
import time
import requests as rest_client
from urllib.parse import urljoin
from fastapi import APIRouter
from starlette.responses import RedirectResponse
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.status import (
HTTP_400_BAD_REQUEST
)
GITLAB_BASE_DOMAIN = 'https://git.youdomain.com'
APP_ID = os.environ.get("APP_ID", None)
APP_SECRET = os.environ.get("APP_SECRET", None)
REDIRECT_URI = os.environ.get("REDIRECT_URI", None)
router = APIRouter()
@router.get("/users/login", tags=["authentication"])
async def login():
unique_state_hash = int(round(time.time() * 1000))
url = f"{GITLAB_BASE_DOMAIN}/oauth/authorize?client_id={APP_ID}&redirect_uri={REDIRECT_URI}&response_type=code&state={unique_state_hash}"
return RedirectResponse(
url=url)
@router.get("/oauth2/callback/gitlab", tags=["authentication"])
async def callback_gitlab(request: Request):
code = request.query_params.get("code", None)
res = rest_client.post(f"{GITLAB_BASE_DOMAIN}/oauth/token", data={'client_id': APP_ID,
'client_secret': APP_SECRET,
'code': code,
'grant_type': 'authorization_code',
'redirect_uri': REDIRECT_URI})
json_res = res.json()
access_token = "access_token"
if access_token not in json_res:
raise HTTPException(
status_code=HTTP_400_BAD_REQUEST,
detail=f"GitLab OAuth2 failure"
)
user_info_request = urljoin(GITLAB_BASE_DOMAIN, f"/api/v4/user?access_token={json_res.get(access_token,'')}")
user_res = rest_client.get(user_info_request)
user_res_data = user_res.json()
print(user_res_data)
return user_res_data