tapdata-sdk 0.3.0

A Python client library for interacting with Tapdata API

Tapdata Python SDK

A Python client library for interacting with Tapdata API.

Features

• 🔐 Complete authentication support
• 📦 Type-safe data models
• 🎯 Clean API interface
• 🔄 Connection and task management
• 📊 Task log queries
• ⚠️ Detailed error handling
• 📝 Comprehensive documentation and type hints

Installation

pip install tapdata-sdk

Or install from source:

git clone https://github.com/lddlww/tapdata_sdk.git
cd tapdata_sdk
pip install -e .

Quick Start

Basic Usage

from tapdata_sdk import TapdataClient

# Initialize client
client = TapdataClient("http://localhost:3030")

# Login
client.login("[e-mail]", "password")

# Query connections
connections = client.connections.list()
for conn in connections:
    print(f"{conn.name}: {conn.status}")

# Query tasks
tasks = client.tasks.list()
for task in tasks:
    print(f"{task.name}: {task.status}")

# Get Table Relation
table_relation = client.tasks.get_table_relation(task_id='xxx')
print(f"{table_relation.table_name_relation}")

Connection Management

from tapdata_sdk import ConnectionType, DatabaseType, Status

# Query source connections
source_connections = client.connections.list_source()

# Query MySQL connections
mysql_connections = client.connections.list_mysql()

# Query valid connections
valid_connections = client.connections.list_valid()

# Filter using enum types
connections = client.connections.list(
    connection_type=ConnectionType.SOURCE,
    database_type=DatabaseType.MYSQL,
    status=Status.COMPLETE
)

# Get single connection details
connection = client.connections.get("connection_id")
print(connection.endpoint)

Task Management

# Query running tasks
running_tasks = client.tasks.list_running()

# Query tasks with specific status
tasks = client.tasks.list(status=Status.RUNNING)

# Get task details
task = client.tasks.get("task_id")

# Start task
client.tasks.start("task_id")

# Stop task
client.tasks.stop("task_id")

# Reset task
client.tasks.reset("task_id")

# Delete task
client.tasks.delete("task_id")

Query Task Logs

import time

# Get logs from the last hour
end_time = int(time.time() * 1000)
start_time = end_time - 3600000  # One hour ago

logs = client.tasks.get_logs(
    task_id="task_id",
    task_record_id="record_id",
    start=start_time,
    end=end_time,
    page=1,
    page_size=20
)

Error Handling

from tapdata_sdk import (
    TapdataError,
    TapdataAuthError,
    TapdataTimeoutError,
    TapdataConnectionError
)

try:
    client.login("[e-mail]", "wrong_password")
except TapdataAuthError as e:
    print(f"Authentication failed: {e.message}")
except TapdataTimeoutError as e:
    print(f"Request timeout: {e.message}")
except TapdataConnectionError as e:
    print(f"Connection error: {e.message}")
except TapdataError as e:
    print(f"API error: {e.message}")

Advanced Configuration

# Custom timeout and SSL verification
client = TapdataClient(
    base_url="https://api.tapdata.io",
    timeout=60,  # 60 second timeout
    verify_ssl=False  # Disable SSL verification (not recommended in production)
)

# Use existing access_token
client = TapdataClient(
    base_url="http://localhost:3030",
    access_token="your-existing-token"
)

# Check authentication status
if client.is_authenticated():
    print("Authenticated")

# Logout
client.logout()

API Reference

TapdataClient

Main client class providing authentication and sub-client access.

Parameters:

• base_url (str): API base URL
• access_token (str, optional): Access token
• timeout (int): Request timeout in seconds, default 30
• verify_ssl (bool): Whether to verify SSL certificate, default True

Methods:

• login(email, password, secret): User login
• logout(): Logout
• is_authenticated(): Check if authenticated
• get_timestamp(): Get server timestamp

Properties:

• connections: ConnectionClient instance
• tasks: TaskClient instance

ConnectionClient

Connection management client.

Methods:

• list(connection_type, database_type, status, skip, limit): Query connection list
• get(connection_id): Get single connection
• list_source(): Get all source connections
• list_target(): Get all target connections
• list_mysql(): Get all MySQL connections
• list_clickhouse(): Get all ClickHouse connections
• list_mongodb(): Get all MongoDB connections
• list_valid(): Get all valid connections
• list_invalid(): Get all invalid connections

TaskClient

Task management client.

Methods:

• list(status, skip, limit): Query task list
• get(task_id): Get single task
• list_running(): Get all running tasks
• start(task_id): Start task
• stop(task_id): Stop task
• reset(task_id): Reset task
• delete(task_id): Delete task
• get_logs(task_id, task_record_id, start, end, page, page_size, levels): Get task logs

Enum Types

from tapdata_sdk import ConnectionType, DatabaseType, Status, LogLevel

# Connection types
ConnectionType.SOURCE
ConnectionType.TARGET

# Database types
DatabaseType.MYSQL
DatabaseType.CLICKHOUSE
DatabaseType.MONGODB
DatabaseType.POSTGRESQL
DatabaseType.ORACLE
DatabaseType.SQLSERVER

# Status
Status.RUNNING
Status.COMPLETE
Status.ERROR
# ... more statuses

# Log levels
LogLevel.INFO
LogLevel.WARN
LogLevel.ERROR
LogLevel.DEBUG

Development

Setup Development Environment

# Clone repository
git clone <repository-url>
cd tapdata-sdk

# Create virtual environment
python -m venv venv
source venv/bin/activate  # Linux/Mac
# or
venv\Scripts\activate  # Windows

# Install dependencies
pip install -e ".[dev]"

Run Tests

pytest tests/

Code Formatting

black tapdata_sdk/
isort tapdata_sdk/

Changelog

v0.3.0 (2026-02-03)

• ✨ added TaskDetail and TaskRelation function

v0.2.0 (2026-01-29)

• ✨ Refactored code architecture with modular design
• 📦 Added data model classes (Connection, Task, TaskLog)
• 🎯 Improved enum types using Python Enum
• 🔧 Optimized error handling with multiple exception types
• 📝 Enhanced documentation and type hints
• 🏗️ Separated client responsibilities (ConnectionClient, TaskClient)
• 🔐 Improved authentication flow
• 📊 Optimized logging

v0.1.0

• 🎉 Initial release

License

MIT License

Contributing

Issues and Pull Requests are welcome!

Support

For questions, please submit an Issue or contact the maintainers.