or run

tessl search
Log in

Version

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/posthog@6.7.x

evals

scenario-1

rubric.jsontask.md
tile.json

tessl/pypi-posthog

tessl install tessl/pypi-posthog@6.7.0

Integrate PostHog into any python application.

Agent Success

Agent success rate when using this tile

89%

Improvement

Agent success rate improvement when using this tile compared to baseline

1.03x

Baseline

Agent success rate without this tile

86%

task.mdevals/scenario-1/

Distributed Feature Flag Cache Provider

Build a custom cache provider for PostHog feature flags that coordinates flag definition fetching across multiple worker processes to minimize API calls.

Requirements

Your implementation should create a cache provider that:

  1. Implements leader election logic so only one worker fetches flag definitions from the PostHog API at a time
  2. Allows multiple workers to read cached flag definitions without redundant API calls
  3. Uses a shared storage mechanism (in-memory dictionary is acceptable for this exercise) to coordinate between workers
  4. Implements proper cache expiration and refresh logic
  5. Provides thread-safe access to cached data

The cache provider must implement these methods:

  • A method to determine if the current worker should fetch new flag definitions (leader election)
  • A method to retrieve cached flag definitions
  • A method to store newly fetched flag definitions for other workers
  • A method to clean up resources on shutdown

Test Cases

  • When multiple workers start simultaneously, only one should be elected to fetch flag definitions @test
  • Workers that are not elected as leader should retrieve definitions from the cache @test
  • Cached flag definitions should expire after a configurable TTL and trigger a new leader election @test
  • The cache provider should handle concurrent access safely from multiple threads @test

Implementation

@generates

API

from typing import Optional, Dict, Any

class FlagDefinitionCacheProvider:
    """
    Cache provider for PostHog feature flag definitions that coordinates
    fetching across multiple workers to minimize API calls.
    """

    def __init__(self, cache_ttl_seconds: int = 60):
        """
        Initialize the cache provider.

        Parameters:
        - cache_ttl_seconds: int - Time-to-live for cached flag definitions in seconds (default: 60)
        """
        pass

    def should_fetch_flag_definitions(self) -> bool:
        """
        Determine if this worker should fetch new flag definitions.
        Implements leader election logic to ensure only one worker fetches at a time.

        Returns:
        bool - True if this worker should fetch, False if it should use cached data
        """
        pass

    def get_flag_definitions(self) -> Optional[Dict[str, Any]]:
        """
        Retrieve cached flag definitions.

        Returns:
        Optional[Dict[str, Any]] - The cached flag definitions, or None if no valid cache exists
        """
        pass

    def on_flag_definitions_received(self, data: Dict[str, Any]) -> None:
        """
        Store newly fetched flag definitions in the cache for other workers to use.

        Parameters:
        - data: Dict[str, Any] - The flag definitions data to cache
        """
        pass

    def shutdown(self) -> None:
        """
        Clean up resources and release any locks held by this cache provider.
        """
        pass

Dependencies { .dependencies }

posthog { .dependency }

Provides feature flag evaluation and management capabilities.

@satisfied-by