tessl install tessl/pypi-posthog@6.7.0Integrate 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%
Build a file-based cache provider for PostHog feature flag definitions that stores flag data in JSON files on the local filesystem. This cache provider should coordinate flag definition fetching across multiple application instances by using file locks to implement leader election.
Your implementation should handle concurrent access from multiple processes and ensure that only one process fetches flag definitions from the PostHog API at a time, while others read from the cached file.
Leader Election: Use file-based locking to determine which process should fetch flag definitions. Only the process that acquires the lock should fetch from the API.
Cache Storage: Store flag definitions as JSON in a file on the filesystem. The cache should persist across application restarts.
Cache Reading: Processes that don't acquire the lock should read flag definitions from the cached file if it exists and is recent enough.
Cache Expiry: Consider cached data stale after 60 seconds. If the cache is stale, the next process to check should attempt to refresh it.
Cleanup: Properly release file locks and clean up resources when the cache provider is shut down.
@generates
import json
import os
import time
from typing import Optional, Dict, Any
import fcntl
class FileFlagCache:
"""
A file-based cache provider for PostHog feature flag definitions.
Uses file locking for leader election to coordinate flag fetching across
multiple processes. Stores flag definitions in a JSON file on the filesystem.
Args:
cache_file_path: Path to the JSON file where flag definitions will be cached
lock_file_path: Path to the lock file used for leader election
cache_ttl_seconds: Time in seconds before cached data is considered stale (default: 60)
"""
def __init__(self, cache_file_path: str, lock_file_path: str, cache_ttl_seconds: int = 60):
pass
def should_fetch_flag_definitions(self) -> bool:
"""
Determines if this process should fetch flag definitions from the API.
Attempts to acquire an exclusive lock on the lock file. If successful,
checks if the cache is stale or missing. Returns True only if this process
has the lock AND the cache needs refreshing.
Returns:
True if this process should fetch flag definitions, False otherwise
"""
pass
def get_flag_definitions(self) -> Optional[Dict[str, Any]]:
"""
Retrieves cached flag definitions from the filesystem.
Reads and parses the JSON cache file. Returns None if the file doesn't
exist or cannot be read.
Returns:
Dictionary containing flag definitions, or None if cache is unavailable
"""
pass
def on_flag_definitions_received(self, data: Dict[str, Any]) -> None:
"""
Stores newly fetched flag definitions to the cache file.
Writes the flag definitions as JSON to the cache file with the current
timestamp. Should be called after successfully fetching from the API.
Args:
data: Dictionary containing flag definitions to cache
"""
pass
def shutdown(self) -> None:
"""
Cleanup method to release locks and close file handles.
Should release any held file locks and properly close all open files.
"""
passProvides the feature flag functionality that this cache provider integrates with.
@satisfied-by