Media asset management for Python, with glue code for various web frameworks
Advanced caching and versioning system for optimizing build performance and managing asset updates, including cache invalidation, version generation, and manifest management.
Cache systems for storing processed assets and improving build performance.
class FilesystemCache:
def __init__(self, directory):
"""
File-based cache implementation.
Parameters:
- directory: Cache directory path
"""
def get(self, key):
"""Get cached value by key."""
def set(self, key, value):
"""Set cached value."""
def has(self, key):
"""Check if key exists in cache."""
def delete(self, key):
"""Delete cached value."""
def clear(self):
"""Clear all cached values."""
class MemoryCache:
def __init__(self, max_size=1000):
"""
In-memory cache implementation.
Parameters:
- max_size: Maximum number of cached items
"""
def get(self, key):
"""Get cached value by key."""
def set(self, key, value):
"""Set cached value."""
def has(self, key):
"""Check if key exists in cache."""
def delete(self, key):
"""Delete cached value."""
def clear(self):
"""Clear all cached values."""
def get_cache(cache_option, env=None):
"""
Create cache instance from configuration.
Parameters:
- cache_option: Cache configuration (str/bool/Cache instance)
- env: Environment for context
Returns:
Cache instance or None
"""Example usage:
from webassets.cache import FilesystemCache, MemoryCache, get_cache
# Filesystem cache
fs_cache = FilesystemCache('./.webassets-cache')
fs_cache.set('bundle_123', compiled_content)
cached = fs_cache.get('bundle_123')
# Memory cache
mem_cache = MemoryCache(max_size=500)
mem_cache.set('filter_abc', processed_data)
# Get cache from configuration
cache = get_cache('filesystem') # Uses default directory
cache = get_cache(True) # Uses default cache
cache = get_cache(False) # No caching
cache = get_cache('memory') # Memory cacheVersion calculation strategies for cache busting and asset update detection.
class Version:
def determine_version(self, bundle, ctx, hunk=None):
"""
Calculate version string for bundle.
Parameters:
- bundle: Bundle instance
- ctx: Version context
- hunk: Optional content hunk
Returns:
Version string
"""
class TimestampVersion(Version):
def determine_version(self, bundle, ctx, hunk=None):
"""Generate version based on file timestamps."""
class HashVersion(Version):
def determine_version(self, bundle, ctx, hunk=None):
"""Generate version based on content hash."""
def get_versioner(versioner_option):
"""
Create versioner from configuration.
Parameters:
- versioner_option: Versioner specification
Returns:
Version instance
"""Example usage:
from webassets.version import HashVersion, TimestampVersion, get_versioner
# Hash-based versioning
hash_versioner = HashVersion()
version = hash_versioner.determine_version(bundle, ctx)
# Timestamp-based versioning
timestamp_versioner = TimestampVersion()
version = timestamp_versioner.determine_version(bundle, ctx)
# Get versioner from config
versioner = get_versioner('hash') # Hash versioning
versioner = get_versioner('timestamp') # Timestamp versioning
versioner = get_versioner(False) # No versioningManifest systems for tracking generated assets and their versions.
class Manifest:
def remember(self, bundle, ctx, version, output_files):
"""
Store bundle information in manifest.
Parameters:
- bundle: Bundle instance
- ctx: Context
- version: Version string
- output_files: Generated output file paths
"""
def query(self, bundle, ctx):
"""
Query manifest for bundle information.
Parameters:
- bundle: Bundle instance
- ctx: Context
Returns:
Stored manifest information or None
"""
class FileManifest(Manifest):
def __init__(self, filename, format='json'):
"""
File-based manifest storage.
Parameters:
- filename: Manifest file path
- format: Storage format ('json', 'yaml')
"""
def get_manifest(manifest_option, env=None):
"""
Create manifest from configuration.
Parameters:
- manifest_option: Manifest specification
- env: Environment for context
Returns:
Manifest instance or None
"""Example usage:
from webassets.version import FileManifest, get_manifest
# JSON manifest
manifest = FileManifest('manifest.json', format='json')
manifest.remember(bundle, ctx, version, ['gen/app-abc123.js'])
info = manifest.query(bundle, ctx)
# Get manifest from config
manifest = get_manifest('json:assets.json') # JSON file
manifest = get_manifest('yaml:assets.yaml') # YAML file
manifest = get_manifest(False) # No manifestException handling for version-related errors.
class VersionIndeterminableError(Exception):
"""Raised when version cannot be determined."""from webassets import Environment
# Basic caching configuration
env = Environment(
'./static',
'/static',
cache=True, # Use default filesystem cache
versions='hash' # Hash-based versioning
)
# Advanced caching configuration
env = Environment(
'./static',
'/static',
cache='filesystem:./.cache', # Custom cache directory
versions='timestamp', # Timestamp versioning
manifest='json:manifest.json' # JSON manifest file
)
# Production configuration
prod_env = Environment(
'./static',
'/static',
debug=False,
cache='filesystem',
versions='hash',
manifest='json:public/assets.json',
auto_build=False
)
# Development configuration
dev_env = Environment(
'./static',
'/static',
debug=True,
cache=False, # No caching in development
versions=False, # No versioning in development
auto_build=True
)from webassets.cache import Cache
import redis
class RedisCache(Cache):
"""Redis-based cache implementation."""
def __init__(self, host='localhost', port=6379, db=0, prefix='webassets:'):
self.redis = redis.Redis(host=host, port=port, db=db)
self.prefix = prefix
def _key(self, key):
return f"{self.prefix}{key}"
def get(self, key):
value = self.redis.get(self._key(key))
return value.decode('utf-8') if value else None
def set(self, key, value):
self.redis.set(self._key(key), value.encode('utf-8'))
def has(self, key):
return self.redis.exists(self._key(key))
def delete(self, key):
self.redis.delete(self._key(key))
def clear(self):
keys = self.redis.keys(f"{self.prefix}*")
if keys:
self.redis.delete(*keys)
# Usage
redis_cache = RedisCache(host='localhost', port=6379)
env = Environment('./static', '/static', cache=redis_cache)from webassets.version import Version
import hashlib
import os
class GitHashVersion(Version):
"""Version based on git commit hash."""
def determine_version(self, bundle, ctx, hunk=None):
import subprocess
try:
# Get current git commit hash
result = subprocess.run(
['git', 'rev-parse', '--short', 'HEAD'],
capture_output=True,
text=True,
cwd=ctx.env.directory
)
if result.returncode == 0:
git_hash = result.stdout.strip()
# Combine with bundle content hash for uniqueness
if hunk:
content_hash = hashlib.md5(hunk.data().encode()).hexdigest()[:8]
return f"{git_hash}-{content_hash}"
else:
return git_hash
except Exception:
pass
# Fallback to timestamp
from webassets.version import TimestampVersion
return TimestampVersion().determine_version(bundle, ctx, hunk)
# Usage
env = Environment('./static', '/static', versions=GitHashVersion())from webassets.version import FileManifest
import json
import os
class CustomManifest(FileManifest):
"""Enhanced manifest with additional metadata."""
def remember(self, bundle, ctx, version, output_files):
# Call parent implementation
super().remember(bundle, ctx, version, output_files)
# Add custom metadata
manifest_data = self._load_manifest()
bundle_key = bundle.resolve_output(ctx)
if bundle_key in manifest_data:
manifest_data[bundle_key].update({
'build_time': datetime.utcnow().isoformat(),
'bundle_id': bundle.id,
'input_files': list(bundle.resolve_contents(ctx)),
'filters': [f.id() for f in bundle.filters],
'file_sizes': {
f: os.path.getsize(os.path.join(ctx.env.directory, f))
for f in output_files
}
})
self._save_manifest(manifest_data)
def _load_manifest(self):
try:
with open(self.filename, 'r') as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
return {}
def _save_manifest(self, data):
with open(self.filename, 'w') as f:
json.dump(data, f, indent=2, sort_keys=True)
# Usage
custom_manifest = CustomManifest('detailed-manifest.json')
env = Environment('./static', '/static', manifest=custom_manifest)from webassets import Environment, Bundle
from webassets.cache import FilesystemCache
# Hierarchical caching
class HierarchicalCache:
"""Multi-level cache with memory and filesystem tiers."""
def __init__(self, memory_size=100, fs_directory='.cache'):
from webassets.cache import MemoryCache, FilesystemCache
self.memory = MemoryCache(max_size=memory_size)
self.filesystem = FilesystemCache(fs_directory)
def get(self, key):
# Try memory first
value = self.memory.get(key)
if value is not None:
return value
# Fall back to filesystem
value = self.filesystem.get(key)
if value is not None:
# Promote to memory
self.memory.set(key, value)
return value
def set(self, key, value):
# Store in both tiers
self.memory.set(key, value)
self.filesystem.set(key, value)
def has(self, key):
return self.memory.has(key) or self.filesystem.has(key)
def delete(self, key):
self.memory.delete(key)
self.filesystem.delete(key)
def clear(self):
self.memory.clear()
self.filesystem.clear()
# Usage
hierarchical_cache = HierarchicalCache(memory_size=200)
env = Environment('./static', '/static', cache=hierarchical_cache)def setup_assets_environment(mode='development'):
"""Setup environment with mode-specific caching and versioning."""
if mode == 'development':
return Environment(
'./src/assets',
'/assets',
debug=True,
cache=False, # No caching in dev
versions=False, # No versioning in dev
auto_build=True
)
elif mode == 'staging':
return Environment(
'./build/assets',
'/assets',
debug=False,
cache='memory', # Fast memory cache
versions='timestamp', # Simple versioning
auto_build=False
)
elif mode == 'production':
return Environment(
'./dist/assets',
'/assets',
debug=False,
cache='filesystem:.cache', # Persistent cache
versions='hash', # Content-based versioning
manifest='json:public/assets.json',
auto_build=False
)
# Usage based on environment
import os
mode = os.environ.get('NODE_ENV', 'development')
env = setup_assets_environment(mode)from webassets import Bundle
# Vendor bundle with stable versioning
vendor_bundle = Bundle(
'vendor/jquery.js',
'vendor/bootstrap.js',
filters='jsmin',
output='gen/vendor-stable.js',
version='vendor-v1.0' # Fixed version for stable vendor code
)
# App bundle with dynamic versioning
app_bundle = Bundle(
'src/app.js',
'src/utils.js',
filters=['babel', 'uglifyjs'],
output='gen/app-%(version)s.js' # Dynamic version placeholder
)
# CSS with hash versioning
css_bundle = Bundle(
'scss/main.scss',
filters=['libsass', 'autoprefixer', 'cssmin'],
output='gen/app-%(version)s.css'
)
env.register('vendor_js', vendor_bundle)
env.register('app_js', app_bundle)
env.register('app_css', css_bundle)Install with Tessl CLI
npx tessl i tessl/pypi-webassets