Skip to content

README.md

Latest commit

 

History

History
226 lines (172 loc) · 5.96 KB

README.md

File metadata and controls

226 lines (172 loc) · 5.96 KB
Raw

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

Supported runtime: Python 3.10 through 3.13. The package publishes both an sdist and a wheel; the repository gate builds and imports both artifacts.

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"
    )

    async def on_transfer_completed(event):
        data = event.get("data", {})
        print(f"Transfer completed: {data.get('filename')}")

    # Listen to transfer events. Frames keep the event payload under data.
    ws.on("transfer.completed", on_transfer_completed)

    # 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

Sessions, Users & Rooms

  • get_sessions() / create_session() - Read or connect the server session
  • ping_session() / disconnect_session() - Maintain or close the session
  • get_session_privileges() - Read current privileges
  • list_users() / get_user() - Read watched users
  • list_rooms() / get_room() / join_room() / leave_room() - Manage rooms

Search

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

Messages

  • list_messages(limit, offset) - List messages
  • get_user_messages(username, limit, offset) - 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

Browse, Events & Administration

  • browse_user() / request_browse(username, folder) - Browse a user's shares
  • get_browse_requests() / respond_to_browse_request() - Manage browse requests
  • get_events(event_type, topic, query, limit, offset) - Read filtered recorded events
  • list_shares() / refresh_shares() - Read or rescan shares
  • get_filters() / update_filters() - Read or update filters
  • get_cache_stats() / invalidate_cache() - Inspect or clear the MediaCore cache

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")

async def on_complete(event):
    print(event.get("data", {}))

ws.on("transfer.completed", on_complete)

Unexpected disconnects are retried with bounded exponential backoff, and the current topic subscriptions are restored after reconnecting. Set ws.max_reconnect_attempts or ws.reconnect_delay when a different retry policy is needed.

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.