포스트

NotionClient

게시 업데이트
By Jeek Lee 9 분읽는 시간

I. NotionClient

NotionClient (그림 I-1) NotionClient(notion-sdk-py)

NotionClient는 JavaScript Notion SDK를 기준으로 작성된 모듈로, Official Notion API의 사용을 좀 더 편하게 사용할 수 있는 기능을 제공한다. 사실 사용하다보면 공식 API에 직접 requests로 보내는 게 나을 수도 있겠다는 생각이 가끔 들긴 하지만, 대부분의 상황에서 복잡한 Notion API의 요청이나 응답 규칙을 하나하나 뜯어보는 것보다 훨씬 나았다.

1

pip install notion-client

II. NotionClient.Client

1
2
3

curl https://api.notion.com/v1/users/01da9b00-e400-4959-91ce-af55307647e5 \
  -H "Authorization: Bearer secret_t1CdN9S8yicG5eWLUOfhcWaOscVnFXns"
  -H "Notion-Version: 2022-06-28"

Notion API 사용을 위해선 위와 같이 Header에 Bearer 토큰을 인증/인가를 위해 사용해야하고, Notion API 버전을 명시해주어야한다. Authorization을 사용하는 건 그렇다 치는데, Notion API 버전 관리가 상당히 귀찮다. Notion API 버전에 따라 요청 양식이 변경되는 경우도 있고, 그럴 때마다 요청을 핸들링하는 코드를 일일이 수정해줘야 한다.

NotionClient의 장점은 Notion API의 변경 사항에 대해 개발자가 조금은 신경을 덜 써도 되는 점에 있는 것 같다. NotionClient에서 Notion API 버전을 관리할 수 있기 때문에, 필요에 따라 Notion API 버전을 자유롭게 변경하고 사용할 수 있다.

1. Client Setting

1
2
3
4
5
6
7

from notion_client import Client
from core.env.setting import NOTION_API_TOKEN


class Notion:
    def __init__(self):
        self.client = Client(auth=NOTION_API_TOKEN)

NotionClient는 위와 같은 방식으로 생성해 사용할 수 있다. Notion API 버전 지정 역시 간단한데, ClientOptions 객체를 통해 지정할 수 있다. 프로젝트에선 NotionClient에서 Default로 지정한 “2022-06-28” 버전을 사용했다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

@dataclass
class ClientOptions:
    """Options to configure the client.

    Attributes:
        auth: Bearer token for authentication. If left undefined, the `auth` parameter
            should be set on each request.
        timeout_ms: Number of milliseconds to wait before emitting a
            `RequestTimeoutError`.
        base_url: The root URL for sending API requests. This can be changed to test with
            a mock server.
        log_level: Verbosity of logs the instance will produce. By default, logs are
            written to `stdout`.
        logger: A custom logger.
        notion_version: Notion version to use.
    """

    auth: Optional[str] = None
    timeout_ms: int = 60_000
    base_url: str = "https://api.notion.com"
    log_level: int = logging.WARNING
    logger: Optional[logging.Logger] = None
    notion_version: str = "2022-06-28"

또한 Client는 Notion의 각 서비스(데이터베이스, 페이지 등) Endpoint에 대응할 수 있도록 구현되어 있다.

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

class BaseClient:
    def __init__(
        self,
        client: Union[httpx.Client, httpx.AsyncClient],
        options: Optional[Union[Dict[str, Any], ClientOptions]] = None,
        **kwargs: Any,
    ) -> None:
        if options is None:
            options = ClientOptions(**kwargs)
        elif isinstance(options, dict):
            options = ClientOptions(**options)

        self.logger = options.logger or make_console_logger()
        self.logger.setLevel(options.log_level)
        self.options = options

        self._clients: List[Union[httpx.Client, httpx.AsyncClient]] = []
        self.client = client

        self.blocks = BlocksEndpoint(self)
        self.databases = DatabasesEndpoint(self)
        self.users = UsersEndpoint(self)
        self.pages = PagesEndpoint(self)
        self.search = SearchEndpoint(self)
        self.comments = CommentsEndpoint(self)

III. Database

NotionDatabase (그림 III-1) NotionDatabase(예시)

Database는 Page의 집합이다. Database의 각 행이 Page이며, Page는 상태, 텍스트, 선택, 파일 등 다양한 속성(property)을 갖는다. query()는 이러한 속성을 바탕으로 Database를 필터링해 Page를 반환한다.

1. client.databases.query()

1
2
3
4
5
6
7
8
9
10
11
12

def query(self, database_id: str, **kwargs: Any) -> SyncAsync[Any]:
    """Get a list of [Pages](https://developers.notion.com/reference/page) contained in the database.

    *[🔗 Endpoint documentation](https://developers.notion.com/reference/post-database-query)*
    """  # noqa: E501
    return self.parent.request(
        path=f"databases/{database_id}/query",
        method="POST",
        query=pick(kwargs, "filter_properties"),
        body=pick(kwargs, "filter", "sorts", "start_cursor", "page_size"),
        auth=kwargs.get("auth"),
    )

query()를 확인해보면, Database ID를 기준으로 다양한 파라미터를 적용할 수 있게 구현되어있다. 공식문서를 기준으로, 각 파라미터의 역할을 확인하면 다음과 같다.

  • filter_properties: 응답에 포함될 속성(property)의 ID
    • 1
2
3
4
5

      # propertyID1, propertyID2 속성만 요청
notion.databases.query({
    database_id,
    filter_properties=["propertyID1", "propertyID2"]
})
  • filter: 응답 필터링 기준
    • 1
2
3
4
5

      # propertyID1의 상태가 시작 전인 Page만 요청
 notion.databases.query({
    database_id,
    filter={'and': [{'property': 'propertyID1', 'status': {'equals': 'Not Started'}}]}
 })
  • sorts: 응답 정렬 기준
    • 1
2
3
4
5

      # Name에 따라 오름차순 정렬
notion.databases.query({
  database_id,
  sorts=[{'property': 'Name', "direction": "ascending"}]
})

2. 예시

query()를 적용한 예시는 다음과 같다.

1
2
3
4

class RequestRepository(metaclass=ABCMeta):
    @abstractmethod
    def find_all_request_application_ready(self):
        pass

1
2
3
4

def find_all_request_application_ready(self) -> List[Request]:
    query = [{'property': 'application/pdf', 'status': {'equals': 'Ready'}}]
    res = self.client.databases.query(NOTION_FILE_REQUEST_DB_ID, filter={'and': query})['results']
    return Request.from_response(res) # Parse JSON Response to object Request

Notion Database가 아닌 다른 Storage(RDS, DynamoDB 등)을 사용하게 될 경우를 대비해, Interface와 Implement 구조를 사용했다.

3. 결과

QueryResult (그림 III-3-1) Notion Database Query Result

Notion API Response가 정상적으로 Object에 매핑되었다.

IV. Page

1. client.pages.retrieve()

1
2
3
4
5
6
7
8
9
10
11

def retrieve(self, page_id: str, **kwargs: Any) -> SyncAsync[Any]:
  """Retrieve a [Page object](https://developers.notion.com/reference/page) using the ID specified.

  *[🔗 Endpoint documentation](https://developers.notion.com/reference/retrieve-a-page)*
  """  # noqa: E501
  return self.parent.request(
      path=f"pages/{page_id}",
      method="GET",
      query=pick(kwargs, "filter_properties"),
      auth=kwargs.get("auth"),
  )

말 그대로 Page 정보를 긁어오는 기능이다. Page의 정보를 업데이트하기 위해선 각 속성(Property)의 정보가 필수적이다. 각 속성의 데이터 구조가 천차만별이기 때문에, 업데이트 전에 Page 정보를 전부 긁어서 필요한 속성만 뽑아내는 것이 정신 건강에 이롭다고 판단했다. 물론 각 속성들을 Object로 매핑해 사용하는 것이 아주 바람직한 방향이지만, 시간 관계상 필요한 속성만 업데이트 하는 방향으로 구현했다.

2. client.pages.update()

1
2
3
4
5
6
7
8
9
10
11

def update(self, page_id: str, **kwargs: Any) -> SyncAsync[Any]:
    """Update [page property values](https://developers.notion.com/reference/page#property-value-object) for the specified page.

    *[🔗 Endpoint documentation](https://developers.notion.com/reference/patch-page)*
    """  # noqa: E501
    return self.parent.request(
        path=f"pages/{page_id}",
        method="PATCH",
        body=pick(kwargs, "archived", "properties", "icon", "cover"),
        auth=kwargs.get("auth"),
    )

파라미터를 통해 Page의 속성뿐만 아니라, 커버 이미지, icon 등을 수정할 수 있다.

3. 예시

1
2
3

@abstractmethod
def update_application_status(self, request_id: str, status: Status):
    pass

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

def update_application_status(self, request_id: str, status: Status):
    application_status_list = self.database_properties['application/pdf']['status']['options']
    status_dict = [status_option for status_option in application_status_list
                   if status_option['name'] == status.value][0]

    page_in_row = self.client.pages.retrieve(request_id)
    properties_new = page_in_row['properties']

    properties_new = {'application/pdf': properties_new.pop('application/pdf')}
    del properties_new['application/pdf']['id']
    del properties_new['application/pdf']['type']

    properties_new['application/pdf']['status'] = status_dict
    self.client.pages.update(page_id=request_id, properties=properties_new)
    return

상태(status)를 업데이트하는 기능이다. Notion API에서 반환하는 상태의 데이터 구조를 살펴보면, 상태 ID, 색상, 이름 등 많은 정보를 복잡하게 담고 있어 다루기 쉽지 않다. 때문에 다음과 같은 방향으로 구현했다.

  1. Database 속성에서, 수정할 속성(application/pdf)의 옵션 중, 이름(name)을 기준으로 옵션 선택
  2. Page ID를 통해 페이지의 전체 정보 요청
  3. Page 전체 정보 중, 수정할 속성 추출(pop)
  4. 수정할 속성 중 업데이트 시 문제가 될 필드 삭제
  5. 수정할 속성에 1번에서 선택한 옵션 적용
  6. 업데이트 요청

분명 더 단순하고 깔끔한 방법이 존재할 것 같지만, 해당 프로젝트는 이 정도 수준에서 구현하는 것으로 마무리했다.

4. 결과

update (그림 IV-4-1) Notion Database Update Result

Notion Database의 Status, message가 업데이트된 결과이다.

V. References

  1. Notion API
  2. NotionClient
FastAPI
fastapi modules notion_client
이 기사는 저작권자의 CC BY 4.0 라이센스를 따릅니다.