Google Sheets 테이블 수집과 셀 갱신

Google Sheets 소스는 Foundry 제공 드라이버로 원천 탐색, 배치 동기화와 증분 수집을 일반 제공(GA)합니다. 대량 읽기는 원천 탐색에서 테이블을 선택하고 배치 동기화를 구성합니다. 특정 스프레드시트 범위를 읽거나 갱신하려면 Sheets API v4를 코드에서 호출합니다.

Data Connection에서 Google Sheets 소스를 생성하고 필요한 인증과 연결 속성을 설정합니다. 기본 커넥터에는 OAuth 대상 accounts.google.com과 www.googleapis.com의 외부 통신이 필요합니다. 사설 경로에는 에이전트 프록시를 사용하며 에이전트와 대상 시스템 양쪽에서 연결을 허용합니다. 에이전트 워커는 권장하지 않는 Legacy 방식입니다.

호출 신원에 따른 소스 선택

서비스 계정의 인증 정보를 호출 사용자 토큰으로 취급하지 않습니다. 함수의 사용자 위임 접근에는 별도의 REST API 소스를 사용합니다.

서비스 계정 파이프라인 준비

Google Sheets 소스의 AuthScheme을 OAuthJWT로 지정하고 Google 서비스 계정 JSON을 구성합니다. 대상 스프레드시트를 서비스 계정과 공유하며 읽기 예제에는 읽기 권한, 쓰기 예제에는 편집 권한이 필요합니다.

변환 저장소에 transforms-external-systems, google-auth, requests, pandas를 추가합니다. 소스의 코드 가져오기를 활성화하여 저장소에 가져옵니다. JSON 인증 정보는 소스의 OAuth_Jwt_Certification 비밀 정보 별칭으로 읽을 수 있어야 합니다.

oauth2.googleapis.com:443과 sheets.googleapis.com:443의 직접 외부 통신 정책을 추가합니다. 소스가 프록시 URI를 제공하면 예제는 토큰 교환과 Sheets 요청 양쪽에 프록시를 적용하며, 없으면 직접 경로를 사용합니다.

사용자 위임 함수 준비

https://sheets.googleapis.com용 REST API 소스를 만들고 OAuth 2.0 외부 인증 애플리케이션을 연결합니다.

accounts.google.com:443, oauth2.googleapis.com:443, sheets.googleapis.com:443을 허용합니다. REST 소스의 내보내기와 코드 가져오기를 활성화하고 Functions 저장소에 가져옵니다.

함수를 부르기 전에 함수 기반 액션 같은 대화형 경로에서 호출 사용자로 앱을 승인해야 합니다. 토큰 갱신이 필요한 함수는 serverless 모드를 사용합니다. deployed 모드는 만료된 호출 사용자 토큰을 자동 갱신하지 못합니다.

TypeScript v1에는 이 워크플로에 필요한 위임 OAuth 인증 정보 접근을 제공하는 예제가 없습니다. 사용자 위임 접근에는 Python 또는 TypeScript v2 함수를 사용합니다.

고정 범위 읽기

소스 RID, 출력 데이터셋 경로, 스프레드시트 ID와 워크시트 이름을 실제 값으로 바꿉니다. 예제 범위는 A1:C3입니다.

Python 파이프라인은 첫 행을 열 이름으로 사용하고 나머지 행을 pandas DataFrame으로 반환하므로, 범위에 헤더와 그 구조에 맞는 값이 있어야 합니다. 함수 예제는 API 응답을 JSON 문자열로 반환합니다. A1 범위 표기는 URL 경로에 넣기 전에 인코딩합니다.

Python 파이프라인

import json
from urllib.parse import quote

import pandas as pd
import requests
from google.auth.transport.requests import Request
from google.oauth2 import service_account
from transforms.api import Output, transform_pandas
from transforms.external.systems import Source, external_systems

SOURCE_RID = "<GOOGLE_SHEETS_SOURCE_RID>"
OUTPUT_DATASET = "<OUTPUT_DATASET_PATH>"
SPREADSHEET_ID = "<SPREADSHEET_ID>"
READ_RANGE = "'<WORKSHEET_NAME>'!A1:C3"


@external_systems(google_sheets=Source(SOURCE_RID))
@transform_pandas(Output(OUTPUT_DATASET))
def compute(google_sheets) -> pd.DataFrame:
    service_account_info = json.loads(
        google_sheets.get_secret("OAuth_Jwt_Certification")
    )
    session = requests.Session()
    proxy_uri = google_sheets.get_https_proxy_uri()
    if proxy_uri:
        session.proxies.update({"http": proxy_uri, "https": proxy_uri})

    credentials = service_account.Credentials.from_service_account_info(
        service_account_info,
        scopes=["https://www.googleapis.com/auth/spreadsheets.readonly"],
    )
    credentials.refresh(Request(session=session))

    response = session.get(
        "https://sheets.googleapis.com/v4/spreadsheets/"
        f"{SPREADSHEET_ID}/values/{quote(READ_RANGE, safe='')}",
        headers={"Authorization": f"Bearer {credentials.token}"},
        timeout=30,
    )
    response.raise_for_status()

    values = response.json()["values"]
    return pd.DataFrame(values[1:], columns=values[0])

Python 함수

import json
from urllib.parse import quote

import requests
from functions.api import function
from functions.sources import get_source

SOURCE_RID = "<GOOGLE_SHEETS_DELEGATED_SOURCE_RID>"
SPREADSHEET_ID = "<SPREADSHEET_ID>"
READ_RANGE = "'<WORKSHEET_NAME>'!A1:C3"


@function(sources=[SOURCE_RID])
def read_google_sheet_values() -> str:
    source = get_source(rid=SOURCE_RID)
    credentials = source.get_session_credentials().get()

    session = requests.Session()
    proxy_uri = source.get_https_proxy_uri()
    if proxy_uri:
        session.proxies.update({"http": proxy_uri, "https": proxy_uri})

    response = session.get(
        "https://sheets.googleapis.com/v4/spreadsheets/"
        f"{SPREADSHEET_ID}/values/{quote(READ_RANGE, safe='')}",
        headers={"Authorization": f"Bearer {credentials.access_token}"},
        timeout=30,
    )
    response.raise_for_status()
    return json.dumps(response.json())

TypeScript v2 함수

import { getFetch, getSource } from "@palantir/functions-sources";

const SOURCE_RID = "<GOOGLE_SHEETS_DELEGATED_SOURCE_RID>";
const SPREADSHEET_ID = "<SPREADSHEET_ID>";
const READ_RANGE = "'<WORKSHEET_NAME>'!A1:C3";

export const config = { sources: [SOURCE_RID] };

export default async function readGoogleSheetValues(): Promise<string> {
  const source = await getSource({ rid: SOURCE_RID });
  const credentials = await source.sessionCredentials?.get();
  if (!credentials || credentials.type !== "oauth") {
    throw new Error("Authorize the outbound application before running this Function");
  }

  const sourceFetch = await getFetch(source);
  const url =
    `https://sheets.googleapis.com/v4/spreadsheets/${SPREADSHEET_ID}/values/` +
    encodeURIComponent(READ_RANGE);
  const response = await sourceFetch(url, {
    headers: { Authorization: `Bearer ${credentials.accessToken}` },
  });
  if (!response.ok) {
    throw new Error(`Google Sheets returned HTTP ${response.status}`);
  }
  return JSON.stringify(await response.json());
}

셀 하나 덮어쓰기

다음 구현은 지정한 워크시트의 B2를 한 번의 spreadsheets.values.update 요청으로 바꿉니다. valueInputOption=RAW로 값을 전달합니다. 대상 셀과 교체값을 확인한 뒤 Python 파이프라인의 CONFIRM_WRITE 또는 함수의 confirm을 활성화해야 합니다.

확인 값이 false이면 Google Sheets에 접속하지 않고 status=dry_run, mutated=false를 반환합니다. true이면 실제 외부 셀을 덮어씁니다. 같은 셀에는 동시에 한 작성자만 접근하도록 하고, 요청 결과가 불확실하면 재시도 전에 셀을 확인합니다.

Python 파이프라인

import json
from urllib.parse import quote

import pandas as pd
import requests
from google.auth.transport.requests import Request
from google.oauth2 import service_account
from transforms.api import Output, transform_pandas
from transforms.external.systems import Source, external_systems

SOURCE_RID = "<GOOGLE_SHEETS_SOURCE_RID>"
OUTPUT_DATASET = "<OUTPUT_DATASET_PATH>"
SPREADSHEET_ID = "<SPREADSHEET_ID>"
TARGET_RANGE = "'<WORKSHEET_NAME>'!B2"
REPLACEMENT_VALUE = "<REPLACEMENT_VALUE>"
CONFIRM_WRITE = False


@external_systems(google_sheets=Source(SOURCE_RID))
@transform_pandas(Output(OUTPUT_DATASET))
def compute(google_sheets) -> pd.DataFrame:
    if not CONFIRM_WRITE:
        return pd.DataFrame([{"status": "dry_run", "mutated": False}])

    service_account_info = json.loads(
        google_sheets.get_secret("OAuth_Jwt_Certification")
    )
    session = requests.Session()
    proxy_uri = google_sheets.get_https_proxy_uri()
    if proxy_uri:
        session.proxies.update({"http": proxy_uri, "https": proxy_uri})

    credentials = service_account.Credentials.from_service_account_info(
        service_account_info,
        scopes=["https://www.googleapis.com/auth/spreadsheets"],
    )
    credentials.refresh(Request(session=session))

    url = (
        "https://sheets.googleapis.com/v4/spreadsheets/"
        f"{SPREADSHEET_ID}/values/{quote(TARGET_RANGE, safe='')}"
        "?valueInputOption=RAW"
    )
    response = session.put(
        url,
        headers={"Authorization": f"Bearer {credentials.token}"},
        json={"range": TARGET_RANGE, "majorDimension": "ROWS", "values": [[REPLACEMENT_VALUE]]},
        timeout=30,
    )
    response.raise_for_status()
    return pd.DataFrame([response.json()])

Python 함수

import json
from urllib.parse import quote

import requests
from functions.api import function
from functions.sources import get_source

SOURCE_RID = "<GOOGLE_SHEETS_DELEGATED_SOURCE_RID>"
SPREADSHEET_ID = "<SPREADSHEET_ID>"
TARGET_RANGE = "'<WORKSHEET_NAME>'!B2"


@function(sources=[SOURCE_RID])
def write_google_sheet_value(new_value: str, confirm: bool = False) -> str:
    if not confirm:
        return json.dumps({"status": "dry_run", "mutated": False})

    source = get_source(rid=SOURCE_RID)
    credentials = source.get_session_credentials().get()

    session = requests.Session()
    proxy_uri = source.get_https_proxy_uri()
    if proxy_uri:
        session.proxies.update({"http": proxy_uri, "https": proxy_uri})

    url = (
        "https://sheets.googleapis.com/v4/spreadsheets/"
        f"{SPREADSHEET_ID}/values/{quote(TARGET_RANGE, safe='')}"
        "?valueInputOption=RAW"
    )
    response = session.put(
        url,
        headers={"Authorization": f"Bearer {credentials.access_token}"},
        json={"range": TARGET_RANGE, "majorDimension": "ROWS", "values": [[new_value]]},
        timeout=30,
    )
    response.raise_for_status()
    return json.dumps(response.json())

TypeScript v2 함수

import { getFetch, getSource } from "@palantir/functions-sources";

const SOURCE_RID = "<GOOGLE_SHEETS_DELEGATED_SOURCE_RID>";
const SPREADSHEET_ID = "<SPREADSHEET_ID>";
const TARGET_RANGE = "'<WORKSHEET_NAME>'!B2";

export const config = { sources: [SOURCE_RID] };

export default async function writeGoogleSheetValue(
  newValue: string,
  confirm: boolean = false,
): Promise<string> {
  if (!confirm) {
    return JSON.stringify({ status: "dry_run", mutated: false });
  }

  const source = await getSource({ rid: SOURCE_RID });
  const credentials = await source.sessionCredentials?.get();
  if (!credentials || credentials.type !== "oauth") {
    throw new Error("Authorize the outbound application before running this Function");
  }

  const sourceFetch = await getFetch(source);
  const url =
    `https://sheets.googleapis.com/v4/spreadsheets/${SPREADSHEET_ID}/values/` +
    `${encodeURIComponent(TARGET_RANGE)}?valueInputOption=RAW`;
  const response = await sourceFetch(url, {
    method: "PUT",
    headers: {
      Authorization: `Bearer ${credentials.accessToken}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      range: TARGET_RANGE,
      majorDimension: "ROWS",
      values: [[newValue]],
    }),
  });
  if (!response.ok) {
    throw new Error(`Google Sheets returned HTTP ${response.status}`);
  }
  return await response.text();
}