View Transformers - Detailed Guide¶
This guide provides detailed examples and implementation patterns for each type of View Transformer.
Registration¶
View Transformers are registered through the ViewTransformerRegistry:
from fastedgy.api_route_model.registry import ViewTransformerRegistry
from fastedgy.dependencies import get_service
# Get the registry service
vtr = get_service(ViewTransformerRegistry)
# Register for specific model
vtr.register_transformer(YourTransformer(), YourModel)
# Register globally (applies to all models)
vtr.register_transformer(GlobalTransformer())
PrePaginateViewTransformer¶
Modifies the database query before data retrieval and pagination.
from fastedgy.api_route_model.view_transformer import PrePaginateViewTransformer
from fastedgy.http import Request
from fastedgy.orm.query import QuerySet, Q
from typing import Any
class QueryOptimizationTransformer(PrePaginateViewTransformer):
"""Optimize queries based on requested fields."""
async def pre_paginate(
self, request: Request, query: QuerySet, ctx: dict[str, Any]
) -> QuerySet:
# Get requested fields from X-Fields header
fields_header = request.headers.get('X-Fields', '')
requested_fields = fields_header.split(',') if fields_header else []
ctx['requested_fields'] = requested_fields
# Optimize query based on requested fields
if any('user.' in field for field in requested_fields):
query = query.select_related('user')
if any('category.' in field for field in requested_fields):
query = query.select_related('category')
if 'tags' in requested_fields:
query = query.prefetch_related('tags')
return query
class ResponseOrderingTransformer(PrePaginateViewTransformer):
"""Apply default ordering when no explicit ordering is requested."""
async def pre_paginate(
self, request: Request, query: QuerySet, ctx: dict[str, Any]
) -> QuerySet:
# Only apply default ordering if no order_by parameter is provided
order_by = request.query_params.get('order_by')
if not order_by:
# Apply default ordering for consistent API responses
query = query.order_by('-created_at', 'id')
return query
GetViewTransformer¶
Transforms individual item dictionaries after serialization.
from fastedgy.api_route_model.view_transformer import GetViewTransformer
from fastedgy.http import Request
from typing import Any
class DataFormattingTransformer(GetViewTransformer):
"""Format data for display purposes."""
async def get_view(
self, request: Request, item, item_dump: dict[str, Any], ctx: dict[str, Any]
) -> dict[str, Any]:
# Format price for display
if 'price' in item_dump and item_dump['price'] is not None:
item_dump['price_formatted'] = f"${float(item_dump['price']):.2f}"
# Format dates for display
if 'created_at' in item_dump and item_dump['created_at']:
from datetime import datetime
if isinstance(item_dump['created_at'], str):
dt = datetime.fromisoformat(item_dump['created_at'].replace('Z', '+00:00'))
item_dump['created_at_formatted'] = dt.strftime('%B %d, %Y')
# Add computed display fields
if 'first_name' in item_dump and 'last_name' in item_dump:
item_dump['full_name'] = f"{item_dump['first_name']} {item_dump['last_name']}"
return item_dump
class DataMaskingTransformer(GetViewTransformer):
"""Mask sensitive data for display purposes."""
async def get_view(
self, request: Request, item, item_dump: dict[str, Any], ctx: dict[str, Any]
) -> dict[str, Any]:
# Mask email addresses for privacy
if 'email' in item_dump and item_dump['email']:
email = item_dump['email']
item_dump['email_masked'] = f"{email[:2]}***@{email.split('@')[1]}"
# Remove internal fields from API responses
internal_fields = ['internal_id', 'debug_info', 'system_notes']
for field in internal_fields:
item_dump.pop(field, None)
return item_dump
GetViewsTransformer¶
Processes collections of items before individual transformation.
from fastedgy.api_route_model.view_transformer import GetViewsTransformer
from fastedgy.http import Request
from typing import Any
class RelatedDataCacheTransformer(GetViewsTransformer):
"""Cache related data to optimize individual item transformations."""
async def get_views(
self, request: Request, items: list, ctx: dict[str, Any]
) -> None:
# Only fetch related data if it will be used in responses
requested_fields = ctx.get('requested_fields', [])
# Cache categories if category data is requested
if any('category' in field for field in requested_fields):
category_ids = [item.category_id for item in items if hasattr(item, 'category_id')]
if category_ids:
categories = await Category.objects.filter(id__in=category_ids).all()
ctx['categories_cache'] = {cat.id: cat for cat in categories}
# Cache user data if user fields are requested
if any('user' in field for field in requested_fields):
user_ids = [item.user_id for item in items if hasattr(item, 'user_id')]
if user_ids:
users = await User.objects.filter(id__in=user_ids).all()
ctx['users_cache'] = {user.id: user for user in users}
PostPaginateViewTransformer¶
Modifies the final pagination response.
from fastedgy.api_route_model.view_transformer import PostPaginateViewTransformer
from fastedgy.http import Request
from fastedgy.schemas.base import Pagination
from typing import Any
class MetadataEnricherTransformer(PostPaginateViewTransformer):
"""Add metadata to pagination response."""
async def post_paginate(
self, request: Request, pagination: Pagination, ctx: dict[str, Any]
) -> None:
# Add response metadata for client information
pagination.metadata = {
'request_timestamp': request.state.start_time if hasattr(request.state, 'start_time') else None,
'response_format': 'paginated',
'fields_selected': bool(request.headers.get('X-Fields')),
'filters_applied': bool(request.headers.get('X-Filter')),
'total_pages': (pagination.total + pagination.limit - 1) // pagination.limit if pagination.limit else 1
}
class AnalyticsTrackingTransformer(PostPaginateViewTransformer):
"""Track API usage for analytics."""
async def post_paginate(
self, request: Request, pagination: Pagination, ctx: dict[str, Any]
) -> None:
# Log API usage (fire and forget)
import asyncio
asyncio.create_task(self._log_usage(request, pagination))
async def _log_usage(self, request: Request, pagination: Pagination):
# Your analytics logging logic here
pass
PreSaveTransformer & PostSaveTransformer¶
Handle data during create/update operations.
from fastedgy.api_route_model.view_transformer import PreSaveTransformer, PostSaveTransformer
from fastedgy.http import Request
from pydantic import BaseModel
from typing import Any
class AuditTransformer(PreSaveTransformer):
"""Add audit fields before saving."""
async def pre_save(
self, request: Request, item, item_data: BaseModel, ctx: dict[str, Any]
) -> None:
user_id = request.headers.get('X-User-ID')
# Set audit fields
if hasattr(item, 'created_by') and not item.created_by:
item.created_by = user_id
if hasattr(item, 'updated_by'):
item.updated_by = user_id
class NotificationTransformer(PostSaveTransformer):
"""Send notifications after successful save."""
async def post_save(
self, request: Request, item, item_data: BaseModel, ctx: dict[str, Any]
) -> None:
# Send notification (async)
import asyncio
asyncio.create_task(self._send_notification(item, ctx))
async def _send_notification(self, item, ctx: dict[str, Any]):
# Your notification logic here
pass
PreDeleteTransformer & PostDeleteTransformer¶
Handle data during delete operations.
from fastedgy.api_route_model.view_transformer import PreDeleteTransformer, PostDeleteTransformer
from fastedgy.http import Request
from fastapi import HTTPException
from typing import Any
class DeleteAuthorizationTransformer(PreDeleteTransformer):
"""Verify user has permission to delete item."""
async def pre_delete(
self, request: Request, item, ctx: dict[str, Any]
) -> None:
user_id = request.headers.get('X-User-ID')
if hasattr(item, 'user_id') and item.user_id != user_id:
raise HTTPException(status_code=403, detail="Not authorized")
class DeleteAuditTransformer(PostDeleteTransformer):
"""Log deletion for audit trail."""
async def post_delete(
self, request: Request, item, ctx: dict[str, Any]
) -> None:
# Log the deletion
print(f"Item {item.id} deleted by user {request.headers.get('X-User-ID')}")
PreUploadTransformer & PostUploadTransformer¶
Handle file uploads with storage control and path manipulation.
from fastedgy.api_route_model.view_transformer import PreUploadTransformer, PostUploadTransformer
from fastedgy.http import Request
from fastapi import UploadFile, HTTPException
from typing import Any
class UploadStorageControlTransformer(PreUploadTransformer):
"""Control storage location based on user permissions."""
async def pre_upload(
self,
request: Request,
record,
field: str,
file: UploadFile,
ctx: dict[str, Any],
) -> bool:
# Validate file type
allowed_types = ['image/jpeg', 'image/png', 'image/webp']
if file.content_type not in allowed_types:
raise HTTPException(status_code=400, detail="File type not allowed")
# Return True for global storage, False for workspace storage
is_admin = request.headers.get('X-User-Role') == 'admin'
return is_admin
class UploadAuditTransformer(PostUploadTransformer):
"""Log upload activity for audit."""
async def post_upload(
self,
request: Request,
record,
field: str,
path: str,
ctx: dict[str, Any],
) -> str:
user_id = request.headers.get('X-User-ID')
print(f"User {user_id} uploaded file to {path}")
return path
PreDownloadTransformer & PostDownloadTransformer¶
Handle file downloads with access control and path resolution.
from fastedgy.api_route_model.view_transformer import PreDownloadTransformer, PostDownloadTransformer
from fastedgy.http import Request
from fastapi import HTTPException
from pathlib import Path
from typing import Any
class DownloadAuthorizationTransformer(PreDownloadTransformer):
"""Verify user has permission to download file."""
async def pre_download(
self,
request: Request,
path: str,
ctx: dict[str, Any],
) -> bool:
workspace_id = request.headers.get('X-Workspace-ID')
# Check if path starts with 'global/' for global storage
if path.startswith('global/'):
return True
# Otherwise require workspace access
if not workspace_id:
raise HTTPException(status_code=403, detail="Workspace required")
return False
class DownloadAuditTransformer(PostDownloadTransformer):
"""Log file downloads for audit."""
async def post_download(
self,
request: Request,
path: str,
served_path: Path,
ctx: dict[str, Any],
) -> Path:
user_id = request.headers.get('X-User-ID')
print(f"User {user_id} downloaded {path}")
return served_path
Context Usage¶
Context dictionary allows sharing data between transformers:
# In PrePaginateViewTransformer - Share data between transformers
ctx['requested_fields'] = request.headers.get('X-Fields', '').split(',')
ctx['response_format'] = request.headers.get('Accept', 'application/json')
# In GetViewTransformer - Use shared context
requested_fields = ctx.get('requested_fields', [])
if 'details' not in requested_fields:
# Skip expensive detail formatting
Best Practices¶
- Performance: Use
GetViewsTransformerfor bulk operations - Context: Share expensive computations via context dictionary
- Error Handling: Use proper HTTP exceptions for client errors
- Async Operations: Use
asyncio.create_task()for fire-and-forget operations - Testing: Test transformers independently with mock requests and contexts
- Registration: Register transformers during app startup, not in route handlers
- File Operations: Return
Truefrom Pre*Transformer for global storage,Falsefor workspace storage - Upload Validation: Validate file type/size in
PreUploadTransformerto fail fast - Download Security: Check permissions in
PreDownloadTransformerbefore file retrieval - Path Manipulation: Use
PostUploadTransformerandPostDownloadTransformerfor path modifications
Next Steps¶
Ready to implement View Transformers in your application?