or run

tessl search
Log in

Version

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/pyrate-limiter@3.9.x

evals

scenario-9

rubric.jsontask.md
tile.json

tessl/pypi-pyrate-limiter

tessl install tessl/pypi-pyrate-limiter@3.9.0

Python Rate-Limiter using Leaky-Bucket Algorithm for controlling request rates in applications with multiple backend storage options.

Agent Success

Agent success rate when using this tile

81%

Improvement

Agent success rate improvement when using this tile compared to baseline

1.45x

Baseline

Agent success rate without this tile

56%

task.mdevals/scenario-9/

JSON File Storage Backend

Build a custom storage backend for a rate limiting system that persists rate limit data to JSON files. The implementation should support basic rate limiting operations with file-based persistence.

Requirements

Your implementation must provide a storage backend that:

  1. Persists rate limit data to a JSON file on disk, allowing rate limits to survive application restarts
  2. Implements all required bucket operations including adding items, removing expired items, counting items, and checking capacity
  3. Supports multiple rate limits simultaneously (e.g., 10 per second AND 100 per minute)
  4. Handles thread safety to prevent data corruption when multiple threads access the same bucket
  5. Provides proper resource cleanup when the bucket is no longer needed

Create a JSONFileBucket class that stores rate items as timestamps in a JSON file, loads existing data on initialization, saves data after modifications, uses locking for thread safety, and implements context manager support.

Capabilities

Bucket Initialization

  • Creating a bucket with a valid file path and rate list initializes correctly and creates the file if it doesn't exist @test

Adding Items to Bucket

  • Adding an item within rate limits succeeds and persists to the JSON file @test
  • Adding an item when rate limit is exceeded returns False @test

Cleaning Up Expired Items

  • Removing expired items (leak operation) cleans up old timestamps correctly based on rate configurations @test

Querying Bucket State

  • Counting items returns the correct number of stored timestamps @test
  • Peeking at a specific index returns the correct item or None if out of bounds @test

Multiple Rate Limit Enforcement

  • Multiple rate limits are enforced correctly (e.g., 5/second and 20/minute both apply) @test

Thread Safety

  • Concurrent operations from multiple threads don't corrupt the data @test

Bucket Cleanup

  • Flushing the bucket removes all items and clears the file @test

Implementation

@generates

API

from typing import List, Optional, Union
from threading import RLock
import json


class JSONFileBucket:
    """
    A file-based storage backend that persists rate limit data to JSON files.

    This bucket stores rate items as timestamps in a JSON file, providing
    persistence across application restarts with thread-safe operations.
    """

    def __init__(self, rates: List, file_path: str):
        """
        Initialize the JSON file bucket.

        Parameters:
        - rates: List of Rate objects defining rate limits
        - file_path: Path to the JSON file for storing data
        """
        pass

    def put(self, item) -> bool:
        """
        Attempt to add an item to the bucket.

        Parameters:
        - item: RateItem to add

        Returns:
        - True if item was added successfully, False if rate limit exceeded
        """
        pass

    def leak(self, current_timestamp: Optional[int] = None) -> int:
        """
        Remove expired items from the bucket.

        Parameters:
        - current_timestamp: Current time in milliseconds (optional)

        Returns:
        - Number of items removed
        """
        pass

    def flush(self) -> None:
        """
        Remove all items from the bucket and clear the file.
        """
        pass

    def count(self) -> int:
        """
        Get the number of items currently in the bucket.

        Returns:
        - Number of items
        """
        pass

    def peek(self, index: int) -> Optional:
        """
        View an item at a specific index without removing it.

        Parameters:
        - index: Index of the item to view

        Returns:
        - RateItem at the index, or None if index is out of bounds
        """
        pass

    def waiting(self, item) -> int:
        """
        Calculate how long to wait (in milliseconds) before the bucket has space.

        Parameters:
        - item: RateItem to check

        Returns:
        - Wait time in milliseconds, or 0 if space is available
        """
        pass

    def close(self) -> None:
        """
        Release any resources held by the bucket.
        """
        pass

    def __enter__(self):
        """
        Enter context manager.

        Returns:
        - self
        """
        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        """
        Exit context manager and cleanup resources.
        """
        pass

Dependencies { .dependencies }

pyrate-limiter { .dependency }

Provides rate limiting functionality and core data structures.

@satisfied-by