tessl/pypi-webassets

Media asset management for Python, with glue code for various web frameworks

Overview

Eval results

Files

Merge System

Name: tessl/pypi-webassets
Author: tessl

The core functionality that manages merging of assets, providing data abstractions (hunks) and filter application tools for the asset processing pipeline. This system handles different types of content sources and enables filter processing with caching support.

Capabilities

Base Hunk

Abstract base class representing a unit of content that can be processed and merged.

class BaseHunk:
    def mtime(self):
        """Return modification time of the content source."""
    
    def id(self):
        """Return unique identifier based on content hash."""
    
    def data(self):
        """Return the content as a string."""
    
    def save(self, filename):
        """Save content to specified file."""
    
    def __eq__(self, other):
        """Compare hunks by content hash."""

File Hunk

Represents content from a single file on the filesystem.

class FileHunk(BaseHunk):
    def __init__(self, filename):
        """
        Initialize with path to source file.
        
        Args:
            filename: Path to the file to read content from
        """
    
    def mtime(self):
        """Return file modification time."""
    
    def data(self):
        """Read and return file contents as UTF-8 string."""

URL Hunk

Represents content from a remote URL with HTTP caching support using etag and last-modified headers.

class UrlHunk(BaseHunk):
    def __init__(self, url, env=None):
        """
        Initialize with URL and optional environment for caching.
        
        Args:
            url: URL to fetch content from
            env: Environment instance for cache access (optional)
        """
    
    def data(self):
        """
        Fetch URL content with HTTP caching support.
        
        Uses environment cache for etag/last-modified headers
        and caches response content and headers.
        """

Memory Hunk

Represents processed content in memory, typically the result of filtering or merging operations.

class MemoryHunk(BaseHunk):
    def __init__(self, data, files=None):
        """
        Initialize with content data and optional source file list.
        
        Args:
            data: String content or file-like object
            files: List of source files this content originated from
        """
    
    def mtime(self):
        """Return None (memory content has no modification time)."""
    
    def data(self):
        """Return content as string, reading from file-like objects if needed."""
    
    def save(self, filename):
        """Save content to file with UTF-8 encoding."""

Merge Function

def merge(hunks, separator=None):
    """
    Merge multiple hunks into a single MemoryHunk.
    
    Args:
        hunks: List of hunk objects to merge
        separator: String to join hunks (default: newline)
        
    Returns:
        MemoryHunk: Combined content from all input hunks
    """

Filter Tool

Applies filters to hunks with caching support, managing the filter execution pipeline and cache operations.

class FilterTool:
    VALID_TRANSFORMS = ('input', 'output')
    VALID_FUNCS = ('open', 'concat')
    
    def __init__(self, cache=None, no_cache_read=False, kwargs=None):
        """
        Initialize filter tool with cache and options.
        
        Args:
            cache: Cache instance for storing filter results
            no_cache_read: Skip cache reads (but still write results)
            kwargs: Default arguments to pass to all filters
        """
    
    def apply(self, hunk, filters, type, kwargs=None):
        """
        Apply filters to a hunk using stream transforms.
        
        Args:
            hunk: Input hunk to process
            filters: List of filter objects to apply
            type: Transform type ('input' or 'output')
            kwargs: Additional arguments for filters
            
        Returns:
            MemoryHunk: Processed content
        """
    
    def apply_func(self, filters, type, args, kwargs=None, cache_key=None):
        """
        Apply filter functions that don't use stream transforms.
        
        Args:
            filters: List of filter objects (only one can have the method)
            type: Function type ('open' or 'concat')
            args: Arguments to pass to filter function
            kwargs: Keyword arguments for filter
            cache_key: Additional cache key components
            
        Returns:
            MemoryHunk: Function result
            
        Raises:
            NoFilters: No filters implement the requested function
            MoreThanOneFilterError: Multiple filters implement the function
        """

Filter Management Functions

def merge_filters(filters1, filters2):
    """
    Merge two filter lists, removing duplicates.
    
    Args:
        filters1: Primary filter list (takes precedence)
        filters2: Secondary filter list (duplicates removed)
        
    Returns:
        list: Combined filter list with filters1 + unique filters from filters2
    """

def select_filters(filters, level):
    """
    Select filters appropriate for given debug level.
    
    Args:
        filters: List of filter objects
        level: Debug level to check against
        
    Returns:
        list: Filters that should run at the specified debug level
    """

Exception Classes

class MoreThanOneFilterError(Exception):
    """
    Raised when multiple filters implement a function that can only be used by one.
    
    Attributes:
        filters: List of conflicting filter objects
    """

class NoFilters(Exception):
    """Raised when no filters implement a requested function."""

Key Features

Content Abstraction

The hunk system provides a unified interface for different content sources:

File-based content: Direct filesystem access with modification tracking
URL-based content: HTTP fetching with caching headers support
Memory content: Processed data from filters or merge operations

HTTP Caching

URL hunks implement sophisticated HTTP caching:

Stores etag and last-modified headers in environment cache
Uses conditional requests (If-None-Match, If-Modified-Since)
Handles 304 Not Modified responses efficiently
Caches both headers and content separately

Filter Pipeline

The FilterTool class manages complex filter execution:

Supports both stream transforms (input/output) and function calls (open/concat)
Implements comprehensive caching with composite cache keys
Handles filter-specific options and additional cache key components
Prevents conflicts when multiple filters implement exclusive functions

Debug Logging

Detailed logging system for troubleshooting:

Uses separate 'webassets.debug' logger controlled by WEBASSETS_DEBUG environment variable
Logs cache hits/misses, filter execution, and hunk operations
Includes content hashes in hunk representations for debugging

Usage Examples

Basic Hunk Operations

from webassets.merge import FileHunk, MemoryHunk, merge

# Create hunks from different sources
file_hunk = FileHunk('style.css')
memory_hunk = MemoryHunk('/* Generated CSS */')

# Merge multiple hunks
combined = merge([file_hunk, memory_hunk], separator='\n\n')

# Save result
combined.save('output.css')

Filter Application

from webassets.merge import FilterTool
from webassets.filter import get_filter

# Setup filter tool with cache
tool = FilterTool(cache=env.cache)

# Apply filters to content
cssmin = get_filter('cssmin')
result = tool.apply(file_hunk, [cssmin], 'output')

URL Content with Caching

from webassets.merge import UrlHunk

# Fetch remote content with caching
url_hunk = UrlHunk('https://cdn.example.com/lib.js', env=env)
content = url_hunk.data()  # Cached on subsequent calls

Install with Tessl CLI