Successfully implemented real-time progress tracking for large batch downloads and longer processes in PyEuropePMC FullTextClient. This professional enhancement provides users with comprehensive progress information during batch operations.
All requested features have been successfully implemented and tested:
download_fulltext_batch method with progress supportclass ProgressInfo:
"""Comprehensive progress tracking for batch operations."""
# Core Properties
total_items: int # Total number of items to process
current_item: int = 0 # Current item being processed
current_pmcid: Optional[str] # Current PMC ID being processed
format_type: str # Download format (pdf, xml, html)
status: str = "pending" # Current status message
# Statistics
successful_downloads: int = 0 # Number of successful downloads
failed_downloads: int = 0 # Number of failed downloads
cache_hits: int = 0 # Number of cache hits
current_file_size: int = 0 # Size of current file in bytes
total_downloaded_bytes: int = 0 # Total bytes downloaded
# Timing
start_time: float # Start timestamp
# Computed Properties
@property
def progress_percent(self) -> float
@property
def elapsed_time(self) -> Optional[float]
@property
def estimated_remaining_time(self) -> Optional[float]
@property
def completion_rate(self) -> float
# Utility Methods
def to_dict(self) -> Dict[str, Any]
def download_fulltext_batch(
self,
pmcids: List[str],
format_type: str = "pdf",
output_dir: Optional[Union[str, Path]] = None,
skip_errors: bool = True,
progress_callback: Optional[Callable[[ProgressInfo], None]] = None, # NEW
progress_update_interval: float = 1.0, # NEW
) -> Dict[str, Optional[Path]]:
New Parameters:
progress_callback: Function called with ProgressInfo updatesprogress_update_interval: Minimum seconds between callback callsfrom pyeuropepmc import FullTextClient, ProgressInfo
def simple_callback(progress: ProgressInfo):
print(f"Progress: {progress.progress_percent:.1f}% - {progress.status}")
client = FullTextClient(enable_cache=True)
results = client.download_fulltext_batch(
pmcids=["PMC1911200", "PMC3312970"],
progress_callback=simple_callback,
progress_update_interval=1.0
)
def detailed_callback(progress: ProgressInfo):
print(f"📊 Progress: {progress.progress_percent:.1f}%")
print(f" Items: {progress.current_item}/{progress.total_items}")
print(f" Success: {progress.successful_downloads}")
print(f" Failed: {progress.failed_downloads}")
print(f" Cache hits: {progress.cache_hits}")
if progress.estimated_remaining_time:
print(f" ETA: {progress.estimated_remaining_time:.1f}s")
from tqdm import tqdm
class ProgressTracker:
def __init__(self):
self.pbar = tqdm(total=100, desc="Download", unit="%")
def update(self, progress: ProgressInfo):
self.pbar.n = progress.progress_percent
self.pbar.set_postfix({
'success': progress.successful_downloads,
'failed': progress.failed_downloads,
'cache': progress.cache_hits
})
self.pbar.refresh()
tracker = ProgressTracker()
results = client.download_fulltext_batch(
pmcids=pmcids,
progress_callback=tracker.update
)
The implementation uses a modular approach with helper methods to maintain code complexity under McCabe limits:
_process_batch_download_item(): Process individual downloads with progress tracking_download_single_by_format(): Handle format-specific downloads_update_progress_after_download(): Update statistics after downloads_handle_batch_download_error(): Handle errors with progress updatesprogress_update_interval prevents excessive callback overheadâś… Progress Callback System: Fully functional
Created comprehensive examples/progress_callbacks_demo.py with:
src/pyeuropepmc/fulltext.py:
src/pyeuropepmc/__init__.py:
examples/progress_callbacks_demo.py: Comprehensive demo script (380 lines)
The progress callback system seamlessly integrates with all existing FullTextClient features:
The progress callback system is production-ready and provides:
Status: âś… FEATURE COMPLETE - Ready for production use