Đồ chơi để thấy GraphQL đỡ phiền

Python Programming Web development Linux

GraphQL là một dạng thiết kế HTTP API được dùng rộng rãi. Mặt lợi của nó thì không cần bàn. Tuy nhiên mình và có thể các bạn cũng thấy hơi phiền vì nó dài dòng. Bài này sẽ giới thiệu một số phương tiện phần mềm để làm việc với nó thoải mái hơn, ở phía client.

Python: Cách dùng Pydantic để kiểm tra hợp lệ

Lấy ví dụ API của BirdWeather. Đây là kho dữ liệu của các trạm thu thập tiếng chim hót, phục vụ nghiên cứu nhận dạng chim từ tiếng hót. Ta sẽ gọi vào truy vấn stations để lấy danh sách các trạm. Một response mẫu sẽ trông như sau:

{
  "data": {
    "stations": {
      "nodes": [
        {
          "id": "4106",
          "name": "   PUC-4106 ",
          "type": "puc",
          "country": "Netherlands",
          "latestDetectionAt": "2025-04-09T17:08:47+02:00"
        },
        {
          "id": "11011",
          "name": "  Lee’s puc   ",
          "type": "puc",
          "country": "Australia",
          "latestDetectionAt": "2025-04-17T06:07:53+10:00"
        }
      ],
      "pageInfo": {
        "endCursor": "Mg",
        "hasNextPage": true,
        "hasPreviousPage": false,
        "startCursor": "MQ"
      },
      "totalCount": 9969
    }
  }
}

Ta thấy response của GraphQL API có đặc điểm là phải đi sâu hai lớp mới tới dữ liệu thực sự ta quan tâm. Cấu trúc của nó luôn có field data, rồi tới field cùng tên với query, rồi mới tới dữ liệu ta cần. Khi nhận dữ liệu từ bên ngoài, như từ API, ta luôn phải làm bước validate (kiểm tra hợp lệ). Dựa vào tài liệu API BirdWeather, ta sẽ định nghĩa model sau với Pydantic:

from datetime import datetime
from enum import StrEnum
from typing import Annotated

from pydantic import BaseModel, Field, ConfigDict
from pydantic.dataclasses import dataclass
from pydantic.alias_generators import to_camel


class StationType(StrEnum):
    BIRDNETPI = 'birdnetpi'
    PUC = 'puc'
    MOBILE = 'mobile'
    YOUTUBE = 'stream_youtube'
    AUDIO = 'stream_audio'


@dataclass(frozen=True, config=ConfigDict(alias_generator=to_camel))
class PageInfo:
    has_next_page: bool
    has_previous_page: bool
    start_cursor: str
    end_cursor: str


class Station(BaseModel):
    id: int
    name: str
    type: StationType
    country: str | None
    latest_detection_at: Annotated[datetime, Field(alias='latestDetectionAt')]


class StationsResponse(BaseModel):
    model_config = ConfigDict(alias_generator=to_camel)

    nodes: tuple[Station, ...]
    page_info: PageInfo
    total_count: int

Ta thấy rằng StationsResponse chỉ mới định nghĩa phần bên trong ['data']['stations'] thôi, chẳng lẽ giờ phải định nghĩa thêm model cho hai lớp bên ngoài nữa sao, nếu vậy, nhân lên với số câu queries thì nhiều quá.

Đừng lo, để ý rằng response của GraphQL luôn có field với tên cố định data, ta sẽ tận dụng Generic trong Python và sự hỗ trợ từ Pydantic để định nghĩa một model khái quát chung:

from typing import Generic, TypeVar

from pydantic import BaseModel

MT = TypeVar('MT')

class GenericResponse(BaseModel, Generic[MT]):
    data: MT

Để sử dụng, ta sẽ truyền vào model cụ thể rồi gọi các hàm model_validate_xxx như thường:

import niquests

def get_stations() -> None:
    post_data = {
        'query': 'query stations ...',
        'variables': {'first': 2},
    }
    try:
        raw = niquests.post(API_URL, json=post_data)
        raw.raise_for_status()
    except niquests.HTTPError as e:
        log.info('Failed to call API. {}', e)
        return
    GenericResponse[StationsResponse].model_validate_json(resp.content)

Tuy nhiên, thực ra model GenericResponse chưa đủ, vì mới giải quyết lớp data. Còn lớp stations hay gì gì nữa. Lớp này hơi tricky tí vì có tên field không cố định, không xài "generic" như trước được. Ta sẽ tạo một model trung gian, on-the-fly, dùng xong rồi vứt, trong cùng một hàm mà ta đang gọi query, vì khi đó ta biết tên field là gì. Để có cách định nghĩa nhanh trong một dòng, ta sẽ tận dụng TypedDict:

from devtools import debug


Wrapper = TypedDict('Wrapper', {'stations': StationsResponse})
resp = GenericResponse[Wrapper].model_validate_json(raw.content or b'')
debug(resp.data['stations'])

Kết quả debug:

resp.data['stations']: StationsResponse(
    nodes=(
        Station(
            id=4106,
            name='   PUC-4106 ',
            type=<StationType.PUC: 'puc'>,
            country='Netherlands',
            latest_detection_at=datetime.datetime(2025, 4, 9, 17, 8, 47, tzinfo=TzInfo(+02:00)),
        ),
        Station(
            id=11011,
            name='  Lee’s puc   ',
            type=<StationType.PUC: 'puc'>,
            country='Australia',
            latest_detection_at=datetime.datetime(2025, 4, 17, 6, 7, 53, tzinfo=TzInfo(+10:00)),
        ),
    ),
    page_info=PageInfo(
        has_next_page=True,
        has_previous_page=False,
        start_cursor='MQ',
        end_cursor='Mg',
    ),
    total_count=9971,
) (StationsResponse)

Trong ví dụ này tôi con dùng một thủ thuật khác: Đổi tên field từ dạng camelCase sang snake_case cho hòa hợp với coding style của Python.

Sau khi validate bằng Pydantic thì trình soạn thảo code sẽ xác định được kiểu biến và sẽ đưa ra gợi ý chính xác:

Zed

Công cụ thử nghiệm API: Nushell

Trước khi viết code ứng dụng, ta thường cần một công cụ để "nghịch" với API, xem nó trả về nội dung thế nào. Thường mọi người dùng Postman cho việc này. Mình thì ưa môi trường dòng lệnh hơn nên dùng công cụ khác. Trước đây mình hay dùng HTTPie nhưng nay thì chuyển qua xài Nushell vì những ưu điểm sẽ kể ra sau đây.

Đây là hình ảnh khi dùng Nushell:

{ query: (open --raw queries/stations.gql), variables: $vars } | http post -t application/json $API_URL

Nu HTTP

Nushell có những tính năng sau, khiến nó rất tiện lợi để khám phá API:

  • Sử dụng dữ liệu có cấu trúc, với các kiểu dữ liệu như record, list, table, nhờ vậy có thể tạo nội dung JSON nhiều tầng dễ dàng bằng cú pháp của Nushell.
  • Response của API được parse thành table và hiển thị thành bảng cho dễ nhìn.

Trong ví dụ trên, mình đã dùng kiểu record của Nushell để tạo nội dung JSON cho request:

{ query: 'query stations ...', variables: { first: 2 } }

Do câu truy vấn GraphQL quá dài nên mình dùng mẹo sau để ra được câu lệnh thật ngắn:

  • Lưu câu vào file stations.gql rồi đọc lên (open --raw queries/stations.gql).
  • Phần cung cấp variables cho GraphQL cũng có thể dài dòng, nên mình lưu vào biến vars rồi sử dụng:
let vars = { first: 2 }
# hoặc
let vars = { first: 2, countries: [Canada Sweden] }

Dữ liệu truyền qua ống | đến lệnh http post thì sẽ được lệnh này dùng làm body cho request. Mình thêm -t application/json để khiến lệnh http chuỗi hóa record kia thành JSON cho hợp lệ. Nếu thấy -t application/json dài dòng thì còn cách khác là chuỗi hóa trước khi truyền cho http post:

❯ { query: (open --raw queries/stations.gql), variables: $vars } | to json | http post  $API_URL

Nushell còn có thể được tận dụng để xử lý hậu kỳ response cho dễ nhìn hơn. Ví dụ ta thấy rằng field latestDetectionAt đang trả về thời gian ở múi giờ không giống ta, khiến khó hình dung, ta có thể xử lý thêm để chuyển đổi múi giờ.

Đầu tiên, để đỡ phải gọi đi gọi lại API khiến server mỏi mệt, ta lưu response vào một biến resp:

Save response

Xử lý bằng lệnh sau:

$resp | update nodes { update latestDetectionAt { into datetime | date to-timezone local | format date } }

Convert timezone

Ở đây có chỗ tricky là phải dùng hai lần lệnh update, do giá trị của field nodes là một danh sách. Lệnh update thứ nhất, update nodes {}, dùng để tính toán giá trị mới cho nodes. Lệnh update nếu áp vào một danh sách, nó sẽ tự động áp closure xử lý lên từng phần tử, và đó là lệnh update thứ hai, update latestDetectionAt.

Giá trị ban đầu của field latestDetectionAt mà API trả về là kiểu string, ta đưa qua into datetime để biến thành kiểu datetime, tiếp đến đưa qua date to-timezone để chuyển đổi múi giờ, và cuối cùng đưa qua format date để format cho dễ nhìn. Nếu không format thì sẽ ra thế này:

Without format date

Nếu ta muốn nhìn thấy chuỗi JSON ban đầu của response, không hiện thành bảng thì sao, thêm --raw vào lệnh http:

❯ { query: (open --raw queries/stations.gql), variables: { first: 4 } } | http post --raw -t application/json $API_URL
{"data":{"stations":{"nodes":[{"id":"4106","name":"   PUC-4106 ","type":"puc","country":"Netherlands","latestDetectionAt":"2025-04-09T17:08:47+02:00"},{"id":"11011","name":"  Lee’s puc   ","type":"puc","country":"Australia","latestDetectionAt":"2025-04-17T06:07:53+10:00"},{"id":"10558","name":"  Vancouver Cambie","type":"puc","country":"Canada","latestDetectionAt":"2025-03-30T16:11:11-07:00"},{"id":"10284","name":" .PUC-10284","type":"puc","country":"United States","latestDetectionAt":"2025-04-16T16:07:12-04:00"}],"pageInfo":{"endCursor":"NA","hasNextPage":true,"hasPreviousPage":false,"startCursor":"MQ"},"totalCount":9991}}}

Chuỗi JSON bị ép lại, khó nhìn quá, muốn format ngay hàng thẳng lối và tô màu như kết quả của HTTPie thì sao? Để hiển thị response ở dạng JSON theo hàng lối thì ta không dùng flag --raw mà chỉ cần chuyển đổi từ table, sau đó lại truyền qua cho Pygment (công cụ bên ngoài, viết bằng Python) để tô màu:

Highlight JSON response

Ta cũng có thể tô màu bằng Shiki (cũng là thư viện tô màu mà blog của mình đang sử dụng), công cụ này phân tích cú pháp chi li hơn Pygment, nhưng cách dùng cũng dài hơn cho chưa hỗ trợ đọc file từ standard input:

Highlight with Shiki

Xong. Mình đã giới thiệu vài món đồ chơi. Bây giờ hãy áp dụng thử nhé!

← Cấu hình autocomplete cho Nushell