urock-transfer 1.3.2

Python bindings for Urock Transfer SDK - reliable data transfer library

Urock Transfer - Python

Python에서 사용할 수 있는 Urock Transfer SDK 바인딩입니다.

설치

pip install urock-transfer

기본 사용법

동기 클라이언트

from urock_transfer import TransferClient, Priority

client = TransferClient()

# 즉시 전송
request_id = client.send(
    data={"event": "click", "value": 123},
    endpoint="https://api.example.com/events",
    priority=Priority.High,
)

# 지연 전송 (큐에 저장)
request_id = client.send(
    data={"event": "click"},
    endpoint="https://api.example.com/events",
    immediate=False,
)

비동기 클라이언트

import asyncio
from urock_transfer import AsyncTransferClient, Priority

async def main():
    client = AsyncTransferClient()

    request_id = await client.send(
        data={"event": "click"},
        endpoint="https://api.example.com/events",
        priority=Priority.High,
    )

asyncio.run(main())

전송 옵션

client.send(
    data={"key": "value"},
    endpoint="https://api.example.com",  # 필수
    priority=Priority.High,               # Low, Normal, High, Critical
    protocol=Protocol.Http,               # Http, WebSocket
    immediate=True,                        # False면 큐에 저장
    retry_count=3,                         # 재시도 횟수
    timeout=30000,                         # 타임아웃 (ms)
    content_encoding="gzip",               # Content-Encoding 헤더 (선택)
)

Content-Encoding 헤더 설정

서버가 압축된 요청을 지원하는 경우, content_encoding 옵션을 사용하여 Content-Encoding 헤더를 설정할 수 있습니다.

from urock_transfer import TransferClient

client = TransferClient()

# Content-Encoding 헤더 설정
client.send(
    data={"large": "data" * 1000},
    endpoint="https://api.example.com",
    content_encoding="gzip",  # Content-Encoding: gzip
)

# Brotli
client.send(
    data={"large": "data" * 1000},
    endpoint="https://api.example.com",
    content_encoding="br",  # Content-Encoding: br
)

참고: 이 옵션은 헤더만 설정합니다. 실제 압축은 애플리케이션에서 미리 수행해야 합니다.

큐 관리

# 백그라운드 큐 시작
client.start_queue(
    interval_ms=5000,   # 처리 간격 (ms)
    batch_size=10,      # 한 번에 처리할 요청 수
    max_retries=3,      # 최대 재시도 횟수
)

# 큐 중지
client.stop_queue()

# 큐 실행 여부 확인
client.is_queue_running()

# 수동으로 한 번 처리
result = client.process_once()
print(result)  # ProcessResult(status='processed', success=5, failed=1, dropped=0)

대기 요청 관리

# 대기 중인 요청 목록
pending = client.list_pending()

# 특정 요청 조회
request = client.load(request_id)

# 특정 요청 삭제
client.delete(request_id)

# 모든 대기 요청 삭제
client.clear_storage()

청크 전송 (대용량 데이터)

# 청크 전송 생성
transfer = client.create_transfer(
    data=large_data,
    endpoint="https://api.example.com/upload",
    chunk_size=65536,
)

# 진행률 콜백
def on_progress(progress):
    print(f"{progress.percent:.1f}% ({progress.sent_chunks}/{progress.total_chunks})")

transfer.on_progress(on_progress)

# 전송 시작
result = transfer.complete()

# 일시정지/재개/취소
transfer.pause()
transfer.resume()
transfer.cancel()

콜백 설정

실패 콜백 (on_fail)

모든 전송 실패 시 호출됩니다.

def on_fail(request, error, attempt):
    endpoint = request.get("options", {}).get("endpoint", "?")
    print(f"[실패] 시도 {attempt}: {endpoint}")
    print(f"  에러: {error}")

client.set_on_fail(on_fail)

삭제 콜백 (on_drop)

max_retries 초과 후 요청이 삭제될 때 호출됩니다.

def on_drop(request, error):
    print(f"요청 삭제됨: {request}")
    print(f"  에러: {error}")

client.set_on_drop(on_drop)

Drop 정책

max_retries 초과 시 요청을 삭제할지 결정합니다.

def drop_policy(request, error):
    # Critical 우선순위는 삭제하지 않음
    priority = request.get("options", {}).get("priority")
    return priority != "Critical"

client.set_drop_policy(drop_policy)

Idle 정책

백그라운드 큐 처리를 진행할지 결정합니다.

def idle_policy(context):
    # 온라인 상태에서만 처리
    status = context.get("network_metrics", {}).get("status")
    return status == "Online"

client.set_idle_policy(idle_policy)

GUI 앱에서 사용 (tkinter)

import tkinter as tk
from urock_transfer import TransferClient

class App:
    def __init__(self, root):
        self.root = root
        self.client = TransferClient()

        # 실패 로그 표시
        def on_fail(request, error, attempt):
            self.root.after(0, lambda: self.log(f"실패: {error}"))

        self.client.set_on_fail(on_fail)

        # 큐 시작
        self.client.start_queue(interval_ms=5000)

    def send_data(self, data):
        self.client.send(
            data=data,
            endpoint="https://api.example.com",
            immediate=False,
        )

    def cleanup(self):
        if self.client.is_queue_running():
            self.client.stop_queue()

타입 및 Enum

from urock_transfer import (
    TransferClient,
    AsyncTransferClient,
    Priority,      # Low, Normal, High, Critical
    Protocol,      # Http, WebSocket
    ProcessResult,
)