Skip to content

Usage Examples

Catalog

Get Catalog List

from kwork_api import KworkClient

async with KworkClient(token="token") as client:
    catalog = await client.catalog.get_list(page=1, category_id=5)

    for kwork in catalog.kworks:
        print(f"{kwork.title}: {kwork.price} RUB")

    # Pagination
    if catalog.pagination:
        print(f"Page {catalog.pagination.current_page} of {catalog.pagination.total_pages}")

Get Kwork Details

details = await client.catalog.get_details(kwork_id=123)

print(f"Title: {details.title}")
print(f"Price: {details.price}")
print(f"Description: {details.full_description}")
print(f"Delivery: {details.delivery_time} days")

Projects

Get Projects List

projects = await client.projects.get_list(page=1)

for project in projects.projects:
    print(f"{project.title} - {project.budget} RUB")

Get Customer Orders

orders = await client.projects.get_payer_orders()

for order in orders:
    print(f"Order #{order.id}: {order.status}")

Get Performer Orders

orders = await client.projects.get_worker_orders()

for order in orders:
    print(f"Work #{order.id}: {order.status}")

User

Get User Info

user_info = await client.user.get_info()
print(f"Username: {user_info.get('username')}")

Get Reviews

reviews = await client.user.get_reviews(page=1)

for review in reviews.reviews:
    print(f"Rating: {review.rating}/5 - {review.comment}")

Get Favorite Kworks

favorites = await client.user.get_favorite_kworks()

for kwork in favorites:
    print(f"Favorite: {kwork.title}")

Reference Data

Get Cities

cities = await client.reference.get_cities()

for city in cities:
    print(f"{city.id}: {city.name}")

Get Countries

countries = await client.reference.get_countries()

for country in countries:
    print(f"{country.id}: {country.name}")

Get Timezones

timezones = await client.reference.get_timezones()

for tz in timezones:
    print(f"{tz.id}: {tz.name} ({tz.offset})")

Notifications

Get Notifications

notifications = await client.notifications.get_list()

for notif in notifications.notifications:
    print(f"{notif.title}: {notif.message}")

print(f"Unread: {notifications.unread_count}")

Fetch New Notifications

new_notifications = await client.notifications.fetch()
print(f"New: {len(new_notifications.notifications)}")

Get Dialogs

dialogs = await client.notifications.get_dialogs()

for dialog in dialogs:
    print(f"Dialog with {dialog.participant.username}: {dialog.last_message}")

Error Handling

from kwork_api import KworkAuthError, KworkApiError, KworkNotFoundError

try:
    catalog = await client.catalog.get_list()
except KworkAuthError as e:
    print(f"Authentication failed: {e}")
except KworkNotFoundError as e:
    print(f"Resource not found: {e}")
except KworkApiError as e:
    print(f"API error [{e.status_code}]: {e.message}")
except Exception as e:
    print(f"Unexpected error: {e}")

Rate Limiting

import asyncio

async def fetch_all_pages():
    all_kworks = []

    for page in range(1, 10):
        try:
            catalog = await client.catalog.get_list(page=page)
            all_kworks.extend(catalog.kworks)

            if not catalog.pagination or not catalog.pagination.has_next:
                break

            # Delay to avoid rate limiting
            await asyncio.sleep(1)

        except KworkRateLimitError:
            print("Rate limited, waiting...")
            await asyncio.sleep(5)

    return all_kworks

Pagination Helper

async def fetch_all_catalog():
    """Fetch all kworks from catalog with pagination."""
    all_kworks = []
    page = 1

    while True:
        catalog = await client.catalog.get_list(page=page)
        all_kworks.extend(catalog.kworks)

        if not catalog.pagination or not catalog.pagination.has_next:
            break

        page += 1
        await asyncio.sleep(0.5)  # Rate limiting

    return all_kworks

More examples in the API Reference.