앞선 과정에서 public API를 등록했습니다. 이제 등록한 API를 이용해서 실제 DB와 상호작용 할 수 있도록 해야 합니다. 단순한 API 키 호출인 프라이빗 API에 비해, 퍼블릭 API는 OAuth 콜백을 통해 사용자의 code를 전달받고, 이를 ACCESS TOKEN으로 전환해서 사용해야 합니다.
토큰을 발급하는 과정을 수행해 보겠습니다.
1. Redirect URI 설정
로컬환경에서 테스트하기 위해 http://localhost:8000/public-api/register 를 redirect URI로 등록합니다. redirect URI는 OAuth를 통한 인증 과정이 끝난 다음, 사용자를 보낼 페이지입니다.
2. 콜백 API 구현
이제, 실제로 OAuth 콜백 호출에 응답하기 위한 콜백 API를 구현합니다. 이 과정에서는 python의 fastapi 라이브러리를 사용합니다. 익숙한 다른 언어/프레임워크를 사용해도 아무 문제 없습니다.
폴더 구조
┌ append/
│ └ append.py # API 기반 페이지 업데이트 코어 로직
├ register/
│ └ register.py # OAuth 콜백 코어 로직
├ .env # 환경변수
└ main.py # FastAPI 함수
폴더 구조에서 append/ 와 register/ 폴더 안에 코어 로직을 분리해 놓은 건 이 프로젝트가 실제로는 AWS Lambda로 배포될 예정이기 때문입니다.
.env
NOTION_CLIENT_ID=#########-####=####-############
NOTION_CLIENT_SECRET=secret_###########################################
REDIRECT_URI=http://localhost:8000/public-api/register
환경 변수를 입력해 줍니다. NOTION_CLIENT_ID 와 NOTION_CLIENT_SECRET 은 API 통합의 OAuth 클라이언트 ID와 OAuth 클라이언트 시크릿을 입력해 줍니다. REDIRECT_URI 역시 앞에서 설정해 준 redirect URI와 일치하도록 입력해줍니다.
main.py
"""
FastAPI Wrapper, 코어 함수들을 로컬 환경에서 실행할 수 있도록 제공합니다
"""
from dotenv import load_dotenv
load_dotenv()
from fastapi import FastAPI, APIRouter, Request
from fastapi.responses import HTMLResponse
app = FastAPI()
public_api_router = APIRouter(prefix="/public-api")
@public_api_router.get("/register")
def register(request: Request):
code = request.query_params.get("code")
access_token = get_access_token(code)
if access_token:
print(access_token) # 배포 시 ACCESS TOKEN 저장 코드로 대체
return {"message": "퍼블릭 API 등록 성공"}
return {"message": "퍼블릭 API 등록 실패"}
app.include_router(daily_quote_router)
public_api_router를 구현해 준 건, 실제 Lambda 배포 시 API Gateway와 형식을 맞추기 위한 것입니다. 로컬 테스트만을 진행하신다면 불필요한 부분입니다. 다만, 등록해 준 redirect URI와 일치시키기 위해 다음과 같이 코드가 수정되어야 합니다.
APIRouter 제외 버전
"""main.py"""
from dotenv import load_dotenv
load_dotenv()
from fastapi import FastAPI, APIRouter, Request
from fastapi.responses import HTMLResponse
app = FastAPI()
@app.get("/public-api/register")
def register(request: Request):
return {"message": "Registration successful"}
@app.post("/public-api/quote")
def quote():
return {"message": "Quote retrieved"}
register.py
"""
Register API Core Logic
Notion Public API를 이용해 사용자 Access Token을 발급받습니다
"""
import requests
import os
NOTION_CLIENT_ID = os.getenv("NOTION_CLIENT_ID")
NOTION_CLIENT_SECRET = os.getenv("NOTION_CLIENT_SECRET")
REDIRECT_URI = os.getenv("REDIRECT_URI")
NOTION_OAUTH_URL = "https://api.notion.com/v1/oauth/token"
def get_access_token(code: str) -> str:
"""OAuth 콜백 함수로서 호출되어 인증 결과를 ACCESS_TOKEN으로 전환합니다"""
headers = {
"Accept": "application/json",
"Content-Type": "application/json",
}
data = {
"grant_type": "authorization_code",
"code": code,
"redirect_uri": REDIRECT_URI,
}
auth = (NOTION_CLIENT_ID, NOTION_CLIENT_SECRET)
res = requests.post(NOTION_OAUTH_URL, data=data, auth=(NOTION_CLIENT_ID, NOTION_CLIENT_SECRET))
if res.status_code != 200:
return ""
token_data = res.json()
access_token = token_data["access_token"]
return access_token
3. 인증 URL 클릭
인증 URL을 이용해 연결 등록을 실행합니다. 인증 URL은 네 개의 쿼리 파라미터를 갖습니다.
- client_id
- response_type
- owner
- redirect_uri
이 중 신경써야 할 것은 redirect_uri 입니다. redirect URI를 여러 개 등록한 경우, 가장 먼저 등록된 redirect URI를 가리키는 인증 URL만 생성됩니다. 다른 redirect URI로 호출하고 싶다면, 이 부분을 바꿔서 링크를 걸어야 합니다.
인증 URL로 접속하면 API 인증 화면이 나타납니다
페이지 선택을 클릭하면, 연결을 추가할 페이지를 선택하는 화면으로 넘어갑니다.
Public API 연동 예제 라는 페이지를 만들어 연결해 보겠습니다. 액세스 허용 을 클릭합니다.
4. notion_client로 페이지 제어
발급받은 토큰을 이용해 액세스 제어를 수행해 보겠습니다. 간단하게 지정된 페이지에 입력받은 텍스트를 저장하는 방식으로 진행합니다.
append.py
"""
Append API Core Logic
notion_client를 이용해 지정된 page에 문구를 삽입합니다.
"""
from notion_client import Client
from notion_client.errors import APIResponseError
def append_block(access_token: str, page_id: str, text: str) -> bool:
"""access_token, page_id, text를 입력받아 지정된 페이지에 텍스트를 삽입합니다"""
try:
client = Client(auth=access_token)
client.blocks.children.append(block_id=page_id, children=[
{
"object": "block",
"type": "paragraph",
"paragraph": {
"rich_text": [
{
"type": "text",
"text": {
"content": text,
}
}
]
}
}
])
return True
except APIResponseError as e:
print(f"노션 API 에러: {e.code} - {e.message}")
return False
except Exception as e:
print(f"노션 API 에러: {e}")
return False
main.py
"""
FastAPI Wrapper, 코어 함수들을 로컬 환경에서 실행할 수 있도록 제공합니다
"""
from dotenv import load_dotenv
load_dotenv()
from fastapi import FastAPI, APIRouter, Request
from fastapi.responses import HTMLResponse
from register.register import get_access_token
from append.append import append_block
app = FastAPI()
public_api_router = APIRouter(prefix="/public-api")
@public_api_router.get("/register")
def register(request: Request):
code = request.query_params.get("code")
access_token = get_access_token(code)
if access_token:
print(access_token) # 배포 시 ACCESS TOKEN 저장 코드로 대체
return {"message": "퍼블릭 API 등록 성공"}
return {"message": "퍼블릭 API 등록 실패"}
@public_api_router.post("/append")
def append(request: Request, access_token: str, page_id: str, text: str):
result = append_block(access_token, page_id, text)
if result:
return {"message": "페이지 삽입 성공"}
return {"message": "페이지 삽입 실패"}
app.include_router(public_api_router)
API로 제어할 준비가 완료되었으니, localhost:8000/docs 로 접속해서 Try it out 버튼을 클릭하고 값들을 입력해 줍니다.
값을 입력하고 Execute 를 클릭해서 실제로 공유된 페이지에 텍스트가 추가된 것을 확인해 봅니다.
이상의 모든 코드는 github에서 참조하실 수 있습니다