Skip to content

README.md

Latest commit

 

History

History
200 lines (152 loc) · 4.56 KB

README.md

File metadata and controls

200 lines (152 loc) · 4.56 KB

slskr Python API Client

Python async client library for the slskr HTTP API. This package is part of the independent slskr project and is not affiliated with or endorsed by Soulseek or its operators.

Features

  • ✅ Full async/await support
  • ✅ Complete HTTP API coverage
  • ✅ Batch operations
  • ✅ WebSocket real-time events
  • ✅ Automatic retries
  • ✅ Comprehensive error handling
  • ✅ Type hints

Installation

pip install slskr-api-client

Quick Start

import asyncio
from slskr import SlskrClient

async def main():
    async with SlskrClient(
        base_url="http://127.0.0.1:5030",
        token="your-bearer-token"
    ) as client:
        # Get stats
        stats = await client.get_stats()
        print(stats)

        # Create search
        search = await client.create_search(query="artist name")
        print(f"Search ID: {search['id']}")

        # Get search results
        results = await client.get_search_details(search['id'])
        print(f"Found {len(results['results'])} results")

asyncio.run(main())

Examples

Batch Operations

from slskr import BatchClient

async def batch_example(client):
    batch = BatchClient(client)

    response = await batch.builder() \
        .get("/api/stats", "stats") \
        .get("/api/transfers", "transfers") \
        .get("/api/messages", "messages") \
        .execute()

    print(f"Completed in {response.total_time_ms}ms")
    print(f"Successful: {len(response.get_successful())}")
    print(f"Failed: {len(response.get_failed())}")

WebSocket Events

from slskr import WebSocketClient

async def websocket_example():
    ws = WebSocketClient(
        base_url="http://127.0.0.1:5030",
        token="your-token"
    )

    # Listen to transfer events
    @ws.on("transfer.completed")
    async def on_transfer_completed(event):
        print(f"Transfer completed: {event['data']}")

    # Connect
    await ws.connect()

    # Subscribe to events
    ws.subscribe("transfer.started", "transfer.completed", "transfer.failed")

    # Keep connection open
    try:
        while True:
            await asyncio.sleep(1)
    finally:
        await ws.disconnect()

Error Handling

from slskr import SlskrClient, ApiError, TimeoutError

async def error_handling():
    async with SlskrClient(...) as client:
        try:
            await client.get_transfer("invalid-id")
        except ApiError as e:
            if e.is_not_found():
                print("Transfer not found")
            elif e.is_unauthorized():
                print("Invalid token")
            elif e.is_conflict():
                print("Transfer conflict")
            else:
                print(f"API error: {e.code}")
        except TimeoutError:
            print("Request timed out")

API Reference

SlskrClient

Main HTTP client for REST API operations.

Methods

Health & Info

  • health() - Server health
  • version() - Version info
  • get_capabilities() - API capabilities
  • get_config() - Configuration
  • get_stats() - Statistics

Search

  • list_searches(limit, offset) - List searches
  • create_search(query, room, target) - Create search
  • get_search_details(search_id) - Search details

Messages

  • list_messages(limit, offset) - List messages
  • get_user_messages(username) - User messages
  • send_message(recipient, content) - Send message
  • acknowledge_message(message_id) - Mark read

Transfers

  • list_transfers(direction, status) - List transfers
  • create_transfer(direction, peer, filename) - Start transfer
  • get_transfer(transfer_id) - Transfer details
  • cancel_transfer(transfer_id) - Cancel transfer

BatchClient

Execute batch operations efficiently.

batch = BatchClient(client)
response = await batch.builder() \
    .get("/api/stats") \
    .post("/api/searches", {"query": "music"}) \
    .delete("/api/transfers/123") \
    .execute()

WebSocketClient

Real-time event streaming.

ws = WebSocketClient(base_url, token)
await ws.connect()
ws.subscribe("transfer.started", "transfer.completed")

@ws.on("transfer.completed")
async def on_complete(event):
    print(event)

Configuration

client = SlskrClient(
    base_url="http://127.0.0.1:5030",
    token="your-token",
    timeout=30,  # Request timeout in seconds
    retries=3,   # Retry attempts
    retry_delay=1,  # Delay between retries
    debug=False  # Enable debug logging
)

Contributing

Contributions welcome! Please see CONTRIBUTING.md

License

AGPL-3.0-only. See the repository LICENSE and NOTICE files for details.