cached_transport

cached_transport

HTTP response caching with smart URL-based invalidation.

This module implements browser-style HTTP caching for Domo API requests with: - Automatic cache invalidation on mutations (POST/PUT/DELETE/PATCH) - Smart URL pattern matching to determine what to invalidate - Collection-level caching for paginated results (looper) - Configurable TTLs per endpoint pattern - Optional custom invalidation rules

Example

from crew_dcs.auth import DomoTokenAuth

Enable caching (default: SMART invalidation strategy)

auth = DomoTokenAuth( ... domo_instance="mycompany", ... domo_access_token="token", ... use_cache=True, ... )

Automatic caching and invalidation

user = await get_user(id="123") # Cached await update_user(id="123") # Auto-invalidates cache user = await get_user(id="123") # Fresh data (cache miss)

CacheEntry dataclass

CacheEntry(
    response: Response,
    cached_at: datetime,
    ttl: int,
    url: str,
)

Cache entry for a single HTTP response.

Attributes:

Name	Type	Description
response	Response	The cached httpx.Response object
cached_at	datetime	Timestamp when response was cached
ttl	int	Time-to-live in seconds
url	str	Original request URL

is_expired

is_expired() -> bool

Check if cache entry has expired.

Source code in src/crew_dcs/client/cached_transport.py

def is_expired(self) -> bool:
    """Check if cache entry has expired."""
    age = datetime.now(UTC) - self.cached_at
    return age.total_seconds() > self.ttl

CachedAsyncHTTPTransport

CachedAsyncHTTPTransport(
    *args,
    cache_size: int = 1000,
    default_ttl: int = DEFAULT_TTL,
    invalidation_strategy: InvalidationStrategy = SMART,
    custom_invalidation_rules: (
        dict[str, list[str]] | None
    ) = None,
    collection_cache_size: int = 100,
    collection_cache_max_records: int = 10000,
    ttl_config: dict[str, int] | None = None,
    collection_ttl_config: dict[str, int] | None = None,
    **kwargs
)

Bases: AsyncHTTPTransport

HTTP transport with response caching and smart invalidation.

This transport layer adds caching to httpx AsyncHTTPTransport with: - Automatic cache invalidation on mutations - Smart URL pattern matching - Separate collection cache for paginated results - Configurable TTLs per endpoint

Parameters:

Name	Type	Description	Default
cache_size	int	Maximum number of cached responses	1000
default_ttl	int	Default TTL in seconds for cached responses	DEFAULT_TTL
invalidation_strategy	InvalidationStrategy	Strategy for cache invalidation	SMART
custom_invalidation_rules	dict[str, list[str]] | None	Optional custom rules for CUSTOM strategy	None
collection_cache_size	int	Maximum number of cached collections	100
collection_cache_max_records	int	Maximum records per collection cache entry	10000
ttl_config	dict[str, int] | None	Optional TTL configuration by URL pattern	None
collection_ttl_config	dict[str, int] | None	Optional collection TTL configuration	None
**kwargs		Additional arguments passed to AsyncHTTPTransport	{}

Example

transport = CachedAsyncHTTPTransport( ... cache_size=1000, ... default_ttl=300, ... invalidation_strategy=InvalidationStrategy.SMART, ... ) client = httpx.AsyncClient(transport=transport)

Source code in src/crew_dcs/client/cached_transport.py

def __init__(
    self,
    *args,
    cache_size: int = 1000,
    default_ttl: int = DEFAULT_TTL,
    invalidation_strategy: InvalidationStrategy = InvalidationStrategy.SMART,
    custom_invalidation_rules: dict[str, list[str]] | None = None,
    collection_cache_size: int = 100,
    collection_cache_max_records: int = 10_000,
    ttl_config: dict[str, int] | None = None,
    collection_ttl_config: dict[str, int] | None = None,
    **kwargs,
):
    super().__init__(*args, **kwargs)

    # Request cache
    self.cache: dict[str, CacheEntry] = {}
    self.cache_size = cache_size
    self.default_ttl = default_ttl
    self.ttl_config = ttl_config or DEFAULT_TTL_CONFIG

    # Collection cache
    self.collection_cache: dict[str, CollectionCacheEntry] = {}
    self.collection_cache_size = collection_cache_size
    self.collection_cache_max_records = collection_cache_max_records
    self.collection_ttl_config = collection_ttl_config or COLLECTION_TTL_CONFIG

    # Invalidator
    self.invalidator = SmartInvalidator(
        strategy=invalidation_strategy,
        custom_rules=custom_invalidation_rules,
    )

    # Statistics
    self.stats = {
        "hits": 0,
        "misses": 0,
        "invalidations": 0,
        "bypasses": 0,
        "collection_hits": 0,
        "collection_misses": 0,
    }

    self._lock = asyncio.Lock()

clear_cache

clear_cache()

Clear all caches.

Source code in src/crew_dcs/client/cached_transport.py

def clear_cache(self):
    """Clear all caches."""
    self.cache.clear()
    self.collection_cache.clear()
    self.stats = {
        "hits": 0,
        "misses": 0,
        "invalidations": 0,
        "bypasses": 0,
        "collection_hits": 0,
        "collection_misses": 0,
    }

get_collection_cache async

get_collection_cache(
    base_url: str, params: dict | None, auth_instance: str
) -> CollectionCacheEntry | None

Retrieve cached collection if not expired.

Parameters:

Name	Type	Description	Default
base_url	str	Base URL	required
params	dict | None	Query parameters	required
auth_instance	str	Authentication instance	required

Returns:

Type	Description
CollectionCacheEntry | None	Cached collection or None

Source code in src/crew_dcs/client/cached_transport.py

async def get_collection_cache(
    self, base_url: str, params: dict | None, auth_instance: str
) -> CollectionCacheEntry | None:
    """Retrieve cached collection if not expired.

    Args:
        base_url: Base URL
        params: Query parameters
        auth_instance: Authentication instance

    Returns:
        Cached collection or None
    """
    key = self.get_collection_cache_key(base_url, params, auth_instance)

    async with self._lock:
        entry = self.collection_cache.get(key)

        if entry:
            if not entry.is_expired():
                self.stats["collection_hits"] += 1
                return entry
            # Remove expired entry
            del self.collection_cache[key]
            self.stats["collection_misses"] += 1

    return None

get_collection_cache_key

get_collection_cache_key(
    base_url: str, params: dict | None, auth_instance: str
) -> str

Generate collection cache key (ignoring pagination params).

Parameters:

Name	Type	Description	Default
base_url	str	Base URL	required
params	dict | None	Query parameters	required
auth_instance	str	Authentication instance	required

Returns:

Type	Description
str	Collection cache key

Source code in src/crew_dcs/client/cached_transport.py

def get_collection_cache_key(
    self, base_url: str, params: dict | None, auth_instance: str
) -> str:
    """Generate collection cache key (ignoring pagination params).

    Args:
        base_url: Base URL
        params: Query parameters
        auth_instance: Authentication instance

    Returns:
        Collection cache key
    """
    parsed = urlparse(base_url)
    path = parsed.path

    # Filter out pagination parameters
    pagination_params = {"offset", "limit", "skip", "top", "page"}
    non_page_params = {}

    if params:
        non_page_params = {
            k: v for k, v in params.items() if k.lower() not in pagination_params
        }

    # Create collection key
    params_str = (
        urlencode(sorted(non_page_params.items())) if non_page_params else ""
    )
    return f"collection:{path}?{params_str}:{auth_instance}"

get_stats

get_stats() -> dict[str, Any]

Get cache statistics.

Returns:

Type	Description
dict[str, Any]	Dictionary with cache statistics

Source code in src/crew_dcs/client/cached_transport.py

def get_stats(self) -> dict[str, Any]:
    """Get cache statistics.

    Returns:
        Dictionary with cache statistics
    """
    total_requests = self.stats["hits"] + self.stats["misses"]
    hit_rate = self.stats["hits"] / total_requests if total_requests > 0 else 0.0

    total_collection_requests = (
        self.stats["collection_hits"] + self.stats["collection_misses"]
    )
    collection_hit_rate = (
        self.stats["collection_hits"] / total_collection_requests
        if total_collection_requests > 0
        else 0.0
    )

    return {
        "request_cache": {
            "total_entries": len(self.cache),
            "max_size": self.cache_size,
            "hits": self.stats["hits"],
            "misses": self.stats["misses"],
            "hit_rate": hit_rate,
            "invalidations": self.stats["invalidations"],
            "bypasses": self.stats["bypasses"],
        },
        "collection_cache": {
            "total_entries": len(self.collection_cache),
            "max_size": self.collection_cache_size,
            "hits": self.stats["collection_hits"],
            "misses": self.stats["collection_misses"],
            "hit_rate": collection_hit_rate,
        },
    }

handle_async_request async

handle_async_request(request: Request) -> Response

Handle HTTP request with caching.

Parameters:

Name	Type	Description	Default
request	Request	The HTTP request to handle	required

Returns:

Type	Description
Response	HTTP response (from cache or server)

Source code in src/crew_dcs/client/cached_transport.py

async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
    """Handle HTTP request with caching.

    Args:
        request: The HTTP request to handle

    Returns:
        HTTP response (from cache or server)
    """
    # Check for cache bypass in extensions
    use_cache = request.extensions.get("use_cache", True)
    invalidate_first = request.extensions.get("invalidate_cache", False)

    # 1. Handle mutations - invalidate related caches BEFORE making request
    if request.method in {"POST", "PUT", "DELETE", "PATCH"}:
        await self._smart_invalidate(request.url)
        use_cache = False  # Never cache mutation responses

    # 2. Manual invalidation
    if invalidate_first:
        await self._invalidate_by_url(str(request.url))

    # 3. Try cache for GET requests
    if use_cache and request.method == "GET":
        cache_key = self._get_cache_key(request)
        if cache_key:
            cached_response = await self._get_from_cache(cache_key)
            if cached_response:
                async with self._lock:
                    self.stats["hits"] += 1
                return cached_response

            async with self._lock:
                self.stats["misses"] += 1
    elif not use_cache:
        async with self._lock:
            self.stats["bypasses"] += 1

    # 4. Make real HTTP request
    response = await super().handle_async_request(request)

    # 5. Cache successful GET responses
    if use_cache and request.method == "GET" and 200 <= response.status_code < 300:
        cache_key = self._get_cache_key(request)
        if cache_key:
            await self._store_in_cache(cache_key, response, str(request.url))

    return response

store_collection_cache async

store_collection_cache(
    base_url: str,
    params: dict | None,
    auth_instance: str,
    data: list,
    request_count: int,
    ttl: int | None = None,
)

Store collection in cache.

Parameters:

Name	Type	Description	Default
base_url	str	Base URL	required
params	dict | None	Query parameters	required
auth_instance	str	Authentication instance	required
data	list	Collection data	required
request_count	int	Number of requests made to fetch collection	required
ttl	int | None	Optional TTL override	None

Source code in src/crew_dcs/client/cached_transport.py

async def store_collection_cache(
    self,
    base_url: str,
    params: dict | None,
    auth_instance: str,
    data: list,
    request_count: int,
    ttl: int | None = None,
):
    """Store collection in cache.

    Args:
        base_url: Base URL
        params: Query parameters
        auth_instance: Authentication instance
        data: Collection data
        request_count: Number of requests made to fetch collection
        ttl: Optional TTL override
    """
    # Don't cache if collection is too large
    if len(data) > self.collection_cache_max_records:
        return

    key = self.get_collection_cache_key(base_url, params, auth_instance)

    if ttl is None:
        # Get TTL from config
        parsed = urlparse(base_url)
        path = parsed.path

        ttl = DEFAULT_COLLECTION_TTL
        for pattern, config_ttl in self.collection_ttl_config.items():
            if re.match(pattern, path):
                ttl = config_ttl
                break

    async with self._lock:
        # Evict oldest if at capacity
        if len(self.collection_cache) >= self.collection_cache_size:
            oldest_key = min(
                self.collection_cache.keys(),
                key=lambda k: self.collection_cache[k].cached_at,
            )
            del self.collection_cache[oldest_key]

        self.collection_cache[key] = CollectionCacheEntry(
            data=data,
            cached_at=datetime.now(UTC),
            ttl=ttl,
            total_records=len(data),
            request_count=request_count,
        )

CollectionCacheEntry dataclass

CollectionCacheEntry(
    data: list,
    cached_at: datetime,
    ttl: int,
    total_records: int,
    request_count: int,
)

Cache entry for complete paginated collections.

Attributes:

Name	Type	Description
data	list	Complete aggregated result from looper
cached_at	datetime	Timestamp when collection was cached
ttl	int	Time-to-live in seconds
total_records	int	Number of records in collection
request_count	int	Number of HTTP requests made to fetch collection

is_expired

is_expired() -> bool

Check if cache entry has expired.

Source code in src/crew_dcs/client/cached_transport.py

def is_expired(self) -> bool:
    """Check if cache entry has expired."""
    age = datetime.now(UTC) - self.cached_at
    return age.total_seconds() > self.ttl

InvalidationStrategy

Bases: Enum

Cache invalidation strategies.

Attributes:

Name	Type	Description
EXACT		Only invalidate exact URL that was mutated
SMART		Invalidate exact URL + parent collections (DEFAULT)
AGGRESSIVE		Invalidate exact + parents + children
CUSTOM		Use custom invalidation rules only

SmartInvalidator

SmartInvalidator(
    strategy: InvalidationStrategy = SMART,
    custom_rules: dict[str, list[str]] | None = None,
)

Smart URL-based cache invalidation without manual rules.

Automatically determines what cache entries to invalidate based on the URL of a mutation request, using intelligent pattern matching.

Parameters:

Name	Type	Description	Default
strategy	InvalidationStrategy	Invalidation strategy to use	SMART
custom_rules	dict[str, list[str]] | None	Optional custom invalidation rules (only used with CUSTOM strategy)	None

Example

invalidator = SmartInvalidator(strategy=InvalidationStrategy.SMART)

Mutation: PUT /api/content/v1/users/123

patterns = invalidator.get_invalidation_patterns( ... "https://domo.com/api/content/v1/users/123" ... )

Returns: ['/api/content/v1/users/123', '/api/content/v1/users', ...]

Source code in src/crew_dcs/client/cached_transport.py

def __init__(
    self,
    strategy: InvalidationStrategy = InvalidationStrategy.SMART,
    custom_rules: dict[str, list[str]] | None = None,
):
    self.strategy = strategy
    self.custom_rules = custom_rules or {}

get_invalidation_patterns

get_invalidation_patterns(mutation_url: str) -> set[str]

Generate cache invalidation patterns for a mutation URL.

Parameters:

Name	Type	Description	Default
mutation_url	str	Full URL of the mutation request	required

Returns:

Type	Description
set[str]	Set of regex patterns that should be invalidated

Source code in src/crew_dcs/client/cached_transport.py

def get_invalidation_patterns(self, mutation_url: str) -> set[str]:
    """Generate cache invalidation patterns for a mutation URL.

    Args:
        mutation_url: Full URL of the mutation request

    Returns:
        Set of regex patterns that should be invalidated
    """
    parsed = urlparse(mutation_url)
    path = parsed.path

    patterns = set()

    # Strategy 1: Exact match (always included)
    patterns.add(self._escape_pattern(path))

    if self.strategy == InvalidationStrategy.EXACT:
        return patterns

    # Strategy 2: Smart (exact + parent collections)
    if self.strategy in (
        InvalidationStrategy.SMART,
        InvalidationStrategy.AGGRESSIVE,
    ):
        parent_patterns = self._extract_parent_patterns(path)
        patterns.update(parent_patterns)

    # Strategy 3: Aggressive (exact + parent + children)
    if self.strategy == InvalidationStrategy.AGGRESSIVE:
        child_pattern = self._escape_pattern(path) + r"/.*"
        patterns.add(child_pattern)

    # Strategy 4: Custom rules (if provided)
    if self.strategy == InvalidationStrategy.CUSTOM and self.custom_rules:
        custom = self._apply_custom_rules(path)
        patterns.update(custom)

    return patterns