/
main.py
75 lines (56 loc) · 2.32 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
import uvicorn
from typing import Union
from fastapi import FastAPI, Request, Depends, HTTPException, Header
from starlette.middleware.cors import CORSMiddleware
from saasus_sdk_python import TenantUserApi
from saasus_sdk_python.callback.callback import Callback
from saasus_sdk_python.middleware.middleware import Authenticate
from saasus_sdk_python.client.client import SignedApiClient
from dotenv import load_dotenv
load_dotenv()
app = FastAPI()
auth = Authenticate()
callback = Callback()
# ApiClientを継承したSignedApiClientを使う
api_client = SignedApiClient()
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# FastAPI用の認証メソッド
def fastapi_auth(request: Request) -> Union[dict, HTTPException]:
auth_header = request.headers.get("Authorization", "")
token = auth_header.replace("Bearer ", "") if "Bearer " in auth_header else ""
referer = request.headers.get("Referer", "")
user_info, error = auth.authenticate(id_token=token, referer=referer)
if error:
raise HTTPException(status_code=401, detail=str(error))
return user_info
# 一時コードを取得する
def get_temp_code(request: Request):
code = request.query_params.get("code")
if not code:
raise HTTPException(status_code=400, detail="code is not provided by query parameter")
return code
@app.get("/credentials")
def get_credentials(request: Request):
return callback.callback_route_function(get_temp_code(request))
@app.get("/userinfo")
def get_user_info(user_info: dict = Depends(fastapi_auth)):
return user_info
@app.get("/users")
def get_tenant_users(auth_user: dict = Depends(fastapi_auth)):
if not auth_user.tenants:
raise HTTPException(status_code=400, detail="No tenants found for the user")
tenant_id = auth_user.tenants[0].id
try:
tenant_user_info = TenantUserApi(api_client=api_client).get_tenant_users(tenant_id=tenant_id,
_headers=api_client.configuration.default_headers)
return tenant_user_info.users
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=80)