User Activity Sorter

A utility for processing and sorting user activity records from async data streams.

Capabilities

Sort user activities by timestamp

The system should process async streams of user activity records and sort them by their timestamp in ascending order.

• Given an async iterable of activities with timestamps [{"user": "alice", "action": "login", "timestamp": 1609459200}, {"user": "bob", "action": "view", "timestamp": 1609455600}], returns them sorted by timestamp in ascending order @test

Sort user activities with custom ordering

The system should support custom sorting logic using key functions that can perform async operations.

• Given an async iterable of activities, sorts them by user name in alphabetical order using a custom key function @test
• Given an async iterable of activities, sorts them in descending order by timestamp when reverse parameter is True @test

Handle empty activity streams

The system should properly handle edge cases with empty data.

• Given an empty async iterable, returns an empty list @test

Implementation

@generates

API

from typing import AsyncIterable, List, Dict, Any, Optional, Callable

async def sort_activities(
    activities: AsyncIterable[Dict[str, Any]],
    key: Optional[Callable[[Dict[str, Any]], Any]] = None,
    reverse: bool = False
) -> List[Dict[str, Any]]:
    """
    Sort user activity records from an async stream.

    Args:
        activities: An async iterable of activity dictionaries
        key: Optional function to extract comparison key from each activity
        reverse: If True, sort in descending order

    Returns:
        A sorted list of activity dictionaries
    """
    pass

Dependencies { .dependencies }

asyncstdlib { .dependency }

Provides async-compatible standard library functions for working with async iterables.

@satisfied-by