Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Callback for progress updates when batching #1430

Closed
alexreg opened this issue Jun 16, 2024 · 2 comments
Closed

Callback for progress updates when batching #1430

alexreg opened this issue Jun 16, 2024 · 2 comments

Comments

@alexreg
Copy link

alexreg commented Jun 16, 2024

What is your feature request?

Any function that takes performs batching on results and automatically aggregates them (i.e., one that takes the container_size parameter) could benefit from firing a caller-specified callback function, which can be used for example to display progress updates to the user. This is particularly useful for calling e.g. section.searchTracks() and similar, which could potentially return many thousands of results.

Are there any workarounds?

None that I'm aware of.

Code Snippets

def show_progress(n_received: int: n_total: Optional[int]) -> None:
    # Note: `n_total` may not always be available (sometimes `None`)
    print(f"Received {n_received} / {n_total} items so far")

section.searchTracks(batch_callback=show_progress)

Additional Context

No response

@Dr-Blank
Copy link
Contributor

Dr-Blank commented Jun 17, 2024

those are just wrappers around fetchItems, which you can call over a loop and show progress.

you can derive from below according to your need. I wrote this to iterate lazily instead of letting it load all at once, you can put you show progress before each yield. with #1373 you can access total items as MediaContainer.totalSize so update you version accordingly.

from typing import Any, Callable, Iterator, TypeVar

from plexapi.exceptions import PlexApiException

T = TypeVar("T")


def plex_batch_iterator(
    func: Callable[..., list[T]], batch_size: int = 50, **kwargs: Any
) -> Iterator[T]:
    """call api in batches and yield results"""
    container_start = 0
    container_size = batch_size
    while True:
        try:
            container = func(
                container_start=container_start,
                maxresults=container_size,
                **kwargs,
            )
            if not container:
                break
            container_start += container_size
            yield from container
        except PlexApiException:
            break

how i used this

...
for artist in plex_batch_iterator(section.searchArtists, sort="titleSort"):  
    ...
...

@alexreg
Copy link
Author

alexreg commented Jun 17, 2024

@Dr-Blank Looks good, thank you. I did think container_start might be the trick, though it's not really documented. I'll give this a try shortly.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Projects
None yet
Development

No branches or pull requests

3 participants