Tableau Hyper API

▶︎ Hyper API

Hyper API

Hyper API는 hyper 파일과의 상호작용을 자동화하여 CRUD 작업을 돕는 API이다.

---

▶︎ Workflow

기본 작업 과정

공식 문서에 의하면, Hyper API는 크게 네 단계를 거쳐 사용된다.

본문에서는 Python을 통해 Hyper API를 사용하는 과정을 다룬다.

 

1. Import

Hyper API 라이브러리를 설치하고 불러오는 과정에 해당한다.

pip install tableauhyperapi

from tableauhyperapi import HyperProcess, Connection, TableDefinition,
SqlType, Telemetry, Inserter, CreateMode, NULLABLE, NOT_NULLABLE,
escape_name, escape_string_literal, TableName

필요에 따라 추가적으로 더 불러올 수 있다.

 

2. Start

HyperProcess를 시작하는 과정에 해당한다.

with HyperProcess(telemetry=Telemetry.SEND_USAGE_DATA_TO_TABLEAU) as hyper:

 

3. Open

Hyper 파일에 연결하는 과정에 해당한다.

with Connection(hyper.endpoint, 'MyExample.hyper', CreateMode.CREATE_AND_REPLACE) as connection:

 

4. Create

이제 작업을 위한 모든 준비가 끝났으므로, CRUD 작업을 수행하면 된다.

---

▶︎ Example

1. Table Definition

sample_table = TableDefinition(
    table_name = "Sample",

    columns = [
        TableDefinition.Column(name="SampleID", type=SqlType.small_int(), nullability=NOT_NULLABLE),
        TableDefinition.Column(name="SampleName", type=SqlType.text(), nullability=NOT_NULLABLE),
        TableDefinition.Column(name="SampleDate", type=SqlType.date(), nullability=NOT_NULLABLE),
        TableDefinition.Column(name="SampleETC", type=SqlType.text(), nullability=NULLABLE)
    ]
)

 

2. Create Table & Insert Rows

from datetime import datetime

def insert_rows():
  path = "/content/sampledata.hyper"

  with HyperProcess(telemetry=Telemetry.SEND_USAGE_DATA_TO_TABLEAU) as hyper:

    with Connection(endpoint=hyper.endpoint,
                    database=path,
                    create_mode=CreateMode.CREATE_AND_REPLACE) as connection:

                    # Create Table
                    connection.catalog.create_table(table_definition=sample_table)

                    sample_data = [
                        [1, "SampleData", datetime(2025, 1, 3), "SampleDescription"],
                        [2, "SampleData2", datetime(2025, 1, 3), "SampleDescription2"]
                    ]

                    # Insert Rows
                    with Inserter(connection, sample_table) as inserter:
                      inserter.add_rows(rows=sample_data)
                      inserter.execute()

 

3. Read Table

from pathlib import Path
import shutil

def read_table():
  path = "/content/sampledata.hyper"

  # 사본 생성
  path_to = Path(shutil.copy(path, "sampledata_read.hyper")).resolve()

  with HyperProcess(telemetry=Telemetry.SEND_USAGE_DATA_TO_TABLEAU) as hyper:

    with Connection(endpoint=hyper.endpoint,
                    database=path_to) as connection:

                    table_name = TableName("Sample")
                    print(f"All Rows in the {table_name}...")

                    rows = connection.execute_list_query(query=f"SELECT * FROM {table_name}")

                    for item in rows:
                      print(item)

 

4. Update Table

from pathlib import Path
import shutil

def update_table():
  path = "/content/sampledata.hyper"

  path_to = Path(shutil.copy(path, "sampledata_updated.hyper")).resolve()

  with HyperProcess(telemetry=Telemetry.SEND_USAGE_DATA_TO_TABLEAU) as hyper:

    with Connection(endpoint=hyper.endpoint,
                    database=path_to) as connection:

                    # Before Update
                    rows_before = connection.execute_list_query(
                        query = f"SELECT {escape_name('SampleID')}, {escape_name('SampleName')}, {escape_name('SampleDate')}, {escape_name('SampleETC')}"
                        f"FROM {escape_name('Sample')}"
                        )

                    print(f"Before Update: 'SampleID', 'SampleName', 'SampleDate', 'SampleETC' "
                    f"columns: {rows_before}\\n")

                    # Updating Process
                    cnt = connection.execute_command(command=f"UPDATE {escape_name('Sample')}"
                    f"SET {escape_name('SampleID')} = {escape_name('SampleID')} + 10")

                    print(f"Updated {cnt} Rows...")

                    # After Update
                    rows_after = connection.execute_list_query(
                        query = f"SELECT {escape_name('SampleID')}, {escape_name('SampleName')}, {escape_name('SampleDate')}, {escape_name('SampleETC')}"
                        f"FROM {escape_name('Sample')}"
                        )

                    print(f"After Update: 'SampleID', 'SampleName', 'SampleDate', 'SampleETC' "
                    f"columns: {rows_after}")

 

5. Delete Rows

from pathlib import Path
import shutil

def delete_rows():
  path = "/content/sampledata.hyper"

  path_to = Path(shutil.copy(path, "sampledata_deleted.hyper"))

  with HyperProcess(telemetry=Telemetry.SEND_USAGE_DATA_TO_TABLEAU) as hyper:

    with Connection(endpoint=hyper.endpoint,
                    database=path_to) as connection:

                    # Delete Rows
                    cnt = connection.execute_command(
                        command=f"DELETE FROM {escape_name('Sample')}"
                        f"WHERE {escape_name('SampleID')} = 1"
                    )

                    print(f"Deleted {cnt} Rows...")

                    rows_after = connection.execute_list_query(
                        query = f"SELECT {escape_name('SampleID')}, {escape_name('SampleName')}, {escape_name('SampleDate')}, {escape_name('SampleETC')}"
                        f"FROM {escape_name('Sample')}"
                        )

                    print(f"columns left: {rows_after}")

 

조금 더 자세하고 다양한 코드는 Hyper API 깃헙에서 확인 가능하다.