feat: add retry utilities, streaming chat, and bulk artifact operations#253
feat: add retry utilities, streaming chat, and bulk artifact operations#253aayushprsingh wants to merge 1 commit intoteng-lin:mainfrom
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds bulk artifact generation and batch download to ArtifactsAPI, a streaming chat generator that yields incremental chunks, and a new async retry utilities module with exponential backoff and rate-limit handling. Changes
Sequence Diagram(s)sequenceDiagram
actor Client
participant ChatAPI
participant HTTPClient
participant Stream as ResponseStream
participant Parser as ChunkParser
Client->>ChatAPI: ask_stream(notebook_id, question, ...)
ChatAPI->>HTTPClient: POST (streaming)
HTTPClient-->>Stream: open streaming response
loop for each line from Stream.aiter_lines()
Stream->>Parser: line
alt parser yields conversation_id
Parser-->>ChatAPI: conversation_id
ChatAPI-->>Client: yield StreamChunk(conversation_id=..., done=False)
end
alt parser yields text chunk
Parser-->>ChatAPI: text
ChatAPI-->>Client: yield StreamChunk(text=..., is_answer=True)
end
alt parser yields reference
Parser-->>ChatAPI: reference
ChatAPI-->>Client: yield StreamChunk(reference=...)
end
end
Stream-->>ChatAPI: stream closed
ChatAPI-->>Client: yield StreamChunk(done=True, conversation_id=last_seen_or_None)
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces bulk artifact operations, a streaming chat interface, and a new retry utility module with exponential backoff. A critical bug was identified in the retry_with_backoff function, which incorrectly attempts to reuse a coroutine object across multiple retry attempts, leading to a RuntimeError. Additionally, feedback was provided to improve API ergonomics in the chat stream by using structured data and to enhance code consistency by replacing os.path operations with pathlib in accordance with PEP 8.
| async def retry_with_backoff( | ||
| coro: Awaitable[T], | ||
| *, | ||
| max_attempts: int = 3, | ||
| initial_delay: float = 1.0, | ||
| max_delay: float = 30.0, | ||
| backoff_factor: float = 2.0, | ||
| jitter: bool = True, | ||
| ) -> T: | ||
| """Simple retry wrapper for a single coroutine. | ||
|
|
||
| Wraps any awaitable and retries it on transient errors with exponential | ||
| backoff. Respects RateLimitError.retry_after when available. | ||
|
|
||
| Args: | ||
| coro: The coroutine to execute and retry. | ||
| max_attempts: Maximum number of attempts (default 3). | ||
| initial_delay: Initial delay in seconds (default 1.0). | ||
| max_delay: Maximum delay in seconds (default 30.0). | ||
| backoff_factor: Exponential backoff multiplier (default 2.0). | ||
| jitter: Add random jitter to delays (default True). | ||
|
|
||
| Returns: | ||
| The result of the coroutine on success. | ||
|
|
||
| Raises: | ||
| The last exception if all attempts fail. | ||
|
|
||
| Example: | ||
| result = await retry_with_backoff( | ||
| client.notebooks.get(notebook_id), | ||
| max_attempts=4, | ||
| ) | ||
| """ | ||
| _retriable: tuple[type[Exception], ...] = ( | ||
| NetworkError, | ||
| RPCTimeoutError, | ||
| RateLimitError, | ||
| ServerError, | ||
| ) | ||
|
|
||
| delay = initial_delay | ||
| last_error: Exception | None = None | ||
|
|
||
| for attempt in range(1, max_attempts + 1): | ||
| try: | ||
| return await coro | ||
| except _retriable as e: | ||
| last_error = e | ||
|
|
||
| if isinstance(e, RateLimitError) and e.retry_after is not None: | ||
| delay = float(e.retry_after) | ||
| elif attempt < max_attempts: | ||
| if jitter: | ||
| delay = min(delay * backoff_factor, max_delay) | ||
| delay *= 0.75 + random.random() * 0.5 | ||
| else: | ||
| delay = min(delay * backoff_factor, max_delay) | ||
|
|
||
| if attempt < max_attempts: | ||
| logger.warning( | ||
| "Retry %s — attempt %d/%d, sleeping %.1fs", | ||
| type(e).__name__, | ||
| attempt, | ||
| max_attempts, | ||
| delay, | ||
| ) | ||
| await asyncio.sleep(delay) | ||
| else: | ||
| logger.error( | ||
| "All %d attempts failed (last: %s)", | ||
| max_attempts, | ||
| e, | ||
| ) | ||
|
|
||
| if last_error is not None: | ||
| raise last_error | ||
| raise RuntimeError(f"All {max_attempts} attempts failed") |
There was a problem hiding this comment.
The retry_with_backoff function is implemented to take an Awaitable (coro) and await it in a loop. However, a coroutine object cannot be awaited more than once. On the second attempt, await coro will raise a RuntimeError: cannot reuse already-awaited coroutine.
To fix this, the function should accept a callable that returns an awaitable (e.g., a lambda or an async function) and call it on each attempt to get a fresh coroutine. Additionally, ensure that tests are updated to reflect this signature change, as per repository guidelines.
async def retry_with_backoff(
coro_fn: Callable[[], Awaitable[T]],
*,
max_attempts: int = 3,
initial_delay: float = 1.0,
max_delay: float = 30.0,
backoff_factor: float = 2.0,
jitter: bool = True,
) -> T:
"""Simple retry wrapper for a single coroutine.
Wraps any awaitable and retries it on transient errors with exponential
backoff. Respects RateLimitError.retry_after when available.
Args:
coro_fn: A callable that returns the coroutine to execute and retry.
max_attempts: Maximum number of attempts (default 3).
initial_delay: Initial delay in seconds (default 1.0).
max_delay: Maximum delay in seconds (default 30.0).
backoff_factor: Exponential backoff multiplier (default 2.0).
jitter: Add random jitter to delays (default True).
Returns:
The result of the coroutine on success.
Raises:
The last exception if all attempts fail.
Example:
result = await retry_with_backoff(
lambda: client.notebooks.get(notebook_id),
max_attempts=4,
)
"""
_retriable: tuple[type[Exception], ...] = (
NetworkError,
RPCTimeoutError,
RateLimitError,
ServerError,
)
delay = initial_delay
last_error: Exception | None = None
for attempt in range(1, max_attempts + 1):
try:
return await coro_fn()
except _retriable as e:
last_error = e
if isinstance(e, RateLimitError) and e.retry_after is not None:
delay = float(e.retry_after)
elif attempt < max_attempts:
if jitter:
delay = min(delay * backoff_factor, max_delay)
delay *= 0.75 + random.random() * 0.5
else:
delay = min(delay * backoff_factor, max_delay)
if attempt < max_attempts:
logger.warning(
"Retry %s — attempt %d/%d, sleeping %.1fs",
type(e).__name__,
attempt,
max_attempts,
delay,
)
await asyncio.sleep(delay)
else:
logger.error(
"All %d attempts failed (last: %s)",
max_attempts,
e,
)
if last_error is not None:
raise last_error
raise RuntimeError(f"All {max_attempts} attempts failed")References
- When a function's signature is updated to return new values, update tests to assert the state of these new returns, including for error handling and early-return paths.
| if isinstance(status, Exception): | ||
| print(f"FAILED {fname}: {status}") | ||
| """ | ||
| import os |
There was a problem hiding this comment.
According to PEP 8, imports should usually be at the top of the file. This local import can be avoided by using pathlib consistently for all path operations in this function, as pathlib is already imported at the file level.
References
- According to PEP 8, imports should usually be at the top of the file. (link)
| ext = extensions.get(artifact_type, "bin") | ||
|
|
||
| specs: list[tuple[str, str]] = [ | ||
| (art.id, os.path.join(output_dir, _safe_filename(art.title or art.id, ext))) |
There was a problem hiding this comment.
For consistency with the use of pathlib elsewhere in this function (e.g., Path(output_dir).mkdir(...)), it's better to use Path objects for path construction instead of os.path.join.
| (art.id, os.path.join(output_dir, _safe_filename(art.title or art.id, ext))) | |
| (art.id, str(Path(output_dir) / _safe_filename(art.title or art.id, ext))) |
| results[os.path.basename(out_path)] = out_path | ||
| except Exception as e: # noqa: BLE001 | ||
| results[os.path.basename(out_path)] = e |
There was a problem hiding this comment.
For consistency with pathlib usage, Path(out_path).name can be used here instead of os.path.basename.
| results[os.path.basename(out_path)] = out_path | |
| except Exception as e: # noqa: BLE001 | |
| results[os.path.basename(out_path)] = e | |
| results[Path(out_path).name] = out_path | |
| except Exception as e: # noqa: BLE001 | |
| results[Path(out_path).name] = e |
References
- For robust file matching, check the filename directly (e.g., path.name) instead of checking for a substring in the full path.
| async def ask_stream( | ||
| self, | ||
| notebook_id: str, | ||
| question: str, | ||
| source_ids: list[str] | None = None, | ||
| conversation_id: str | None = None, | ||
| ) -> AsyncIterator[tuple[str | None, bool, list[ChatReference] | None, str | None, bool]]: |
There was a problem hiding this comment.
The current implementation of ask_stream yields a raw tuple, which can be hard for consumers to work with as they need to remember the order and types of the elements. The example in the pull request summary suggests a more ergonomic API that yields an object with attributes (chunk.text, chunk.done), which is much more readable and less error-prone.
Consider defining a dataclass for the stream chunks to improve API usability. This would align the implementation with the better design showcased in the PR summary.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 11
🧹 Nitpick comments (2)
src/notebooklm/_chat.py (2)
188-225: Consider using a NamedTuple for the yield type.The 5-tuple
(text, is_answer, references, conversation_id, done)requires callers to remember positional semantics. ANamedTuplewould improve readability and enable attribute access:class StreamChunk(NamedTuple): text: str | None is_answer: bool references: list[ChatReference] | None conversation_id: str | None done: boolThis allows
chunk.textinstead ofchunk[0], and the example in the docstring would still work with tuple unpacking.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/notebooklm/_chat.py` around lines 188 - 225, Introduce a NamedTuple (e.g., StreamChunk) to replace the raw 5-tuple returned by ask_stream so callers can use attribute access; define StreamChunk with fields text: str | None, is_answer: bool, references: list[ChatReference] | None, conversation_id: str | None, done: bool and update the ask_stream signature and its yields to return StreamChunk instances instead of plain tuples, adjust the docstring example to show both attribute access (chunk.text) and that tuple unpacking still works, and update any internal references to the tuple structure accordingly.
226-274: Extract duplicated request-building logic into a shared helper.Lines 226-274 nearly duplicate the request construction in
ask()(lines 96-145). If the API request format changes, both methods must be updated in lockstep.Consider extracting a private method like
_build_chat_request(notebook_id, question, source_ids, conversation_id)that returns(body, url, conversation_id).♻️ Sketch of refactored helper
async def _build_chat_request( self, notebook_id: str, question: str, source_ids: list[str] | None, conversation_id: str | None, ) -> tuple[str, str, str, bool]: """Build request body and URL for chat API calls. Returns: Tuple of (body, url, conversation_id, is_new_conversation). """ if source_ids is None: source_ids = await self._core.get_source_ids(notebook_id) is_new_conversation = conversation_id is None if is_new_conversation: conversation_id = str(uuid.uuid4()) conversation_history = None else: assert conversation_id is not None conversation_history = self._build_conversation_history(conversation_id) # ... rest of request building ... return body, url, conversation_id, is_new_conversation🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/notebooklm/_chat.py` around lines 226 - 274, Duplicate request-building logic in ask() and the block around params_json/f_req_json/encoded_req/body_parts/body/url should be extracted into a private helper (e.g. _build_chat_request) that accepts (notebook_id, question, source_ids, conversation_id) and returns (body, url, conversation_id, is_new_conversation); the helper should call self._core.get_source_ids when source_ids is None, use self._build_conversation_history when conversation_id is provided, build params -> params_json -> f_req_json -> encoded_req, assemble body_parts (including CSRF token from self._core.auth.csrf_token), compute url with _reqid and self._core.auth.session_id, and the caller (ask and the other method) should call this helper and use the returned body, url and conversation_id instead of duplicating the construction logic.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/notebooklm/_artifacts.py`:
- Around line 2604-2615: The extension map currently hardcodes
ArtifactType.SLIDE_DECK to "pdf", causing mismatched filename suffixes when
users request output_format="pptx"; update the mapping logic for SLIDE_DECK (and
the duplicated mapping near lines referenced around 2654–2656) to use
output_format if provided (e.g., use output_format or "pdf") so ext =
extensions.get(artifact_type, "bin") will yield the requested slide format;
ensure this change is applied in the _download_by_type handling where
artifact_type and ext are used.
- Around line 2481-2488: Reject non-positive concurrency immediately in the
public generator helpers: in async def generate_bulk(...) validate
max_concurrency at the top and raise a clear ValueError if max_concurrency <= 0
(e.g., "max_concurrency must be > 0"), and apply the same precondition check to
the companion public helper defined later (the other generate_* method around
the 2551-2559 region) so you never create asyncio.Semaphore(0) and avoid hanging
workers.
- Line 2670: The code raises ArtifactError in the branch that handles unknown
artifact types (raise ArtifactError(f"No download method for artifact type:
{artifact_type}")) but ArtifactError is not defined; add a proper reference to
the defined exception by importing or defining it near the top of
src/notebooklm/_artifacts.py (e.g., import ArtifactError from the module that
declares it) so the unknown-type branch raises the intended user-facing
exception instead of causing a NameError; update imports to include
ArtifactError and keep the existing raise statement unchanged.
- Around line 2600-2629: The current _safe_filename/stem logic can produce
duplicate paths for multiple artifacts with identical or similar titles causing
races in specs/_download_one; make the filename unique before scheduling by
incorporating a unique identifier (e.g., art.id or an incremental counter) into
the generated filename/stem (modify the specs list comprehension or
_safe_filename usage) so each tuple in specs is guaranteed distinct, and ensure
results keys (os.path.basename(out_path)) remain stable; update anywhere specs
is created and where _safe_filename is called to use the uniqueized stem.
- Around line 2527-2548: generate_bulk currently treats returned
GenerationStatus objects as successes; update _generate_one to treat
GenerationStatus(status="failed", ...) as a failure path: after awaiting
generate_fn(nb_id), if the result is a GenerationStatus with status == "failed"
then either raise (when error_mode == "fail_fast") or store the GenerationStatus
in results as an error (so return_exceptions/collect_all sees it as failure).
Refer to _generate_one, generate_fn, generate_bulk, GenerationStatus and results
in the diff and ensure the success counting and logger.info use isinstance(r,
GenerationStatus) and r.status == "succeeded" (or inverse) so failed
GenerationStatus values are not counted as successes.
- Around line 2483-2484: The parameter annotation notebook_ids: list[str] is
shadowing the class method list(), causing A003; change the annotation to
builtins.list[str] (and add an import for builtins at the top if not present) so
the type refers to the builtin list rather than the class method; update any
other annotations in this class that use list[...] the same way (e.g.,
function/method signatures around generate_fn and the class that defines
list()).
In `@src/notebooklm/_chat.py`:
- Around line 233-234: The else-branch that calls
_build_conversation_history(conversation_id) lacks the same type-narrowing
assertion used in ask(), so add an assertion like assert conversation_id is not
None before calling self._build_conversation_history to satisfy mypy and ensure
conversation_id is treated as non-None in that branch (referencing the method
_build_conversation_history and the ask() pattern).
- Around line 285-309: ask_stream() is streaming pieces of an answer but never
caches the assembled conversation turn, so follow-up queries lose context;
initialize accumulated_text = "" before the async loop, append each chunk's text
(from _extract_answer_and_refs_from_chunk) to accumulated_text when text is
yielded, and before emitting the final tuple with done=True call
self._core.cache_conversation_turn(conversation_id=server_conv_id,
user_question=question, assistant_answer=accumulated_text) (or the equivalent
cache_conversation_turn signature used elsewhere) so
_build_conversation_history() will find the turn for subsequent questions.
In `@src/notebooklm/_retry.py`:
- Around line 170-180: The example usage of RetryManager omits waiting for
backoff: after catching a RateLimitError you must call
retry.on_transient_error(e) and then await retry.sleep_before_retry() (or check
the boolean return from on_transient_error and only await sleep_before_retry()
when it indicates a retry is planned) before the next loop iteration; update the
example loop near the RetryManager usage to await sleep_before_retry() (and
honor retry.attempt < retry.max_attempts and retry.complete()) so the backoff is
actually applied.
- Around line 121-128: The backoff logic currently multiplies delay before
sleeping and applies jitter after clamping, causing the first retry to wait
initial_delay*backoff_factor and allowing jitter to exceed max_delay; fix the
three implementations (the block in this file plus
RetryManager.on_transient_error and retry_with_backoff) so you sleep the current
delay first, then update delay for the next attempt by setting delay = min(delay
* backoff_factor, max_delay); compute jitter by applying the random multiplier
to the current sleep value and then clamp that jittered sleep to max_delay
before sleeping (i.e., sleep = min(jittered_sleep, max_delay)), ensuring the
first retry uses initial_delay and jitter never pushes the sleep above
max_delay.
- Around line 197-202: The RetryManager.__aenter__ currently returns self
without resetting internal state, so reuse yields stale attempt, _current_delay,
and _done; update __aenter__ to reinitialize self.attempt = 0,
self._current_delay = self.initial_delay (or the original initial_delay value),
and self._done = False before returning self, or alternatively document and
enforce that RetryManager is single-use (e.g., raise if attempt or _done
indicate prior use) so callers cannot accidentally reuse an exhausted instance;
refer to RetryManager, __aenter__, attempt, _current_delay, _done, and
initial_delay when making the change.
---
Nitpick comments:
In `@src/notebooklm/_chat.py`:
- Around line 188-225: Introduce a NamedTuple (e.g., StreamChunk) to replace the
raw 5-tuple returned by ask_stream so callers can use attribute access; define
StreamChunk with fields text: str | None, is_answer: bool, references:
list[ChatReference] | None, conversation_id: str | None, done: bool and update
the ask_stream signature and its yields to return StreamChunk instances instead
of plain tuples, adjust the docstring example to show both attribute access
(chunk.text) and that tuple unpacking still works, and update any internal
references to the tuple structure accordingly.
- Around line 226-274: Duplicate request-building logic in ask() and the block
around params_json/f_req_json/encoded_req/body_parts/body/url should be
extracted into a private helper (e.g. _build_chat_request) that accepts
(notebook_id, question, source_ids, conversation_id) and returns (body, url,
conversation_id, is_new_conversation); the helper should call
self._core.get_source_ids when source_ids is None, use
self._build_conversation_history when conversation_id is provided, build params
-> params_json -> f_req_json -> encoded_req, assemble body_parts (including CSRF
token from self._core.auth.csrf_token), compute url with _reqid and
self._core.auth.session_id, and the caller (ask and the other method) should
call this helper and use the returned body, url and conversation_id instead of
duplicating the construction logic.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 233e7f89-b935-4e80-b93f-357cd4cb9655
📒 Files selected for processing (3)
src/notebooklm/_artifacts.pysrc/notebooklm/_chat.pysrc/notebooklm/_retry.py
| notebook_ids: list[str], | ||
| generate_fn: Callable[[str], Coroutine[Any, Any, "GenerationStatus"]], |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# First, check file size and get line counts around the key areas
wc -l src/notebooklm/_artifacts.pyRepository: teng-lin/notebooklm-py
Length of output: 1895
🏁 Script executed:
# Read the area around Line 249 to find the list method definition
sed -n '245,255p' src/notebooklm/_artifacts.pyRepository: teng-lin/notebooklm-py
Length of output: 1895
🏁 Script executed:
# Read the area around Lines 2483-2484 to see the exact annotation
sed -n '2480,2490p' src/notebooklm/_artifacts.pyRepository: teng-lin/notebooklm-py
Length of output: 1895
🏁 Script executed:
# Also check the full method signature and context for the function containing lines 2483-2484
sed -n '2475,2495p' src/notebooklm/_artifacts.pyRepository: teng-lin/notebooklm-py
Length of output: 1895
🏁 Script executed:
# Look for import statements and how list/annotations are used elsewhere in the file
head -50 src/notebooklm/_artifacts.pyRepository: teng-lin/notebooklm-py
Length of output: 1895
🌐 Web query:
Python type annotation list[str] shadowed builtin in class scope
💡 Result:
Yes, using list[str] (or any builtin type like list, dict, str, etc.) in a type annotation within a Python class body can fail if a prior class attribute, method, or variable named list has been defined earlier in the class scope. This shadows the builtin list type. This occurs because Python class definitions create a temporary local namespace (scope) during execution. Names defined in this namespace (e.g., methods or class variables) shadow outer scopes, including builtins, for subsequent name lookups within the same class body. Type annotations like list[str] are evaluated in this local class scope at definition time (for runtime) or analyzed similarly by static type checkers like mypy and Pylance. Example causing the issue: class Foo: def list(self) -> None: # Defines 'list' as a method pass def bar(self, items: list[str]) -> None: # 'list' resolves to the method above, not builtin list pass This leads to runtime TypeError: 'function' object is not subscriptable, or mypy error: "Function ... is not valid as a type". Official Python behavior (from docs.python.org/3/reference/executionmodel.html): "A class definition is an executable statement... The namespace of the class definition becomes the attribute dictionary of the class." Name resolution follows normal rules, with the class namespace as the local scope. mypy confirms: "The list type annotation... resolves to the symbol you're defining in the class body" (mypy issue #17517). Workarounds: - Rename the shadowing attribute/method (recommended). - Use builtins.list[str] after import builtins. - Define a type alias before the shadowing name: _List = list, then _List[str]. - Linters like Ruff (rule A003) warn about this. This is standard Python scoping, not specific to typing (post-Python 3.9+ built-in generics). Avoid shadowing builtins in classes, especially with type annotations.
Citations:
- 1: Class with a classmethod named
listfails when other methods uselistin annotations python/mypy#17517 - 2: https://stackoverflow.com/questions/78101227/why-i-cannot-type-a-list-parameter-in-a-class-with-an-existing-list-method
- 3: https://docs.astral.sh/ruff/rules/builtin-attribute-shadowing/
- 4: https://stackoverflow.com/questions/69104779/mypy-list-shadowing/69105927
- 5: https://docs.python.org/3.15/tutorial/classes.html
- 6: http://docs.python.org/3.11/reference/executionmodel.html
- 7: https://docs.python.org/3/reference/executionmodel.html
- 8: https://peps.python.org/pep-0484/
🌐 Web query:
Ruff A003 rule builtin shadowing annotations
💡 Result:
The Ruff rule A003, named "builtin-attribute-shadowing", detects class attributes and methods that shadow Python builtins when those shadowed names are referenced within the class scope. This is particularly relevant for type annotations, where using a shadowed builtin like list[int] might resolve to the class method instead of the built-in type. Example triggering code: class Class: @staticmethod def list -> None: pass @staticmethod def repeat(value: int, times: int) -> list[int]: return [value] * times Here, list[int] in the annotation refers to the list method, not the built-in list. The rule was refined over time: initially broader, but updated (e.g., via fixes in 2024-2025) to only flag actual shadowing usages within the class, such as in annotations, decorators, argument defaults, and attribute initializers. Methods with @typing.override or @typing_extensions.override are ignored. Builtins can be ignored via lint.flake8-builtins.ignorelist in configuration. Recent issues (as of 2026) discuss diagnostic improvements and edge cases like property setters.
Citations:
- 1: https://docs.astral.sh/ruff/rules/builtin-attribute-shadowing/
- 2: https://beta.ruff.rs/docs/rules/builtin-attribute-shadowing/
- 3: A003 misses some attributes referenced within the class scope astral-sh/ruff#20171
- 4: Remove rule A003 astral-sh/ruff#7806
- 5: Rule
A003intended behaviour astral-sh/ruff#23074 - 6: https://docs.astral.sh/ruff/rules/
🌐 Web query:
Python __future__ annotations shadowing builtins class method
💡 Result:
No, from future import annotations does not cause shadowing of the builtin classmethod (or other builtins like list, str) by a class method with the same name in type annotations. Class methods defined in a class namespace shadow outer names (including builtins) during class body execution due to Python's scoping rules. This happens regardless of the annotations future import. Without postponed annotations (default behavior pre-3.10 or without import): - Annotations are eagerly evaluated during class body execution. - A method named e.g. list shadows builtins.list, so -> list[int] in another method tries to subscript the method object, causing TypeError: 'function' object is not subscriptable. With from future import annotations (postponed annotations, PEP 563): - Annotations are stored as strings in annotations, avoiding eager evaluation and NameErrors for forward refs. - However, when later resolved (e.g. via typing.get_type_hints or annotationlib.get_annotations in 3.14+), evaluation uses the class namespace first, then globals, then builtins. - The shadowing method is still found first, so 'list[int]' resolves to the method, not builtins.list. Tools like mypy may flag "Function is not valid as a type". This is expected Python behavior, not specific to the future import. Workarounds: - Rename the shadowing method. - Use builtins.classmethod, builtins.list, etc. in annotations. - Define type aliases outside the class, e.g. _List = list. In Python 3.14+, deferred evaluation (PEP 649/749) has similar class-namespace-first resolution, but from future import annotations continues stringified behavior (deprecated long-term). Example: from future import annotations class Foo: @classmethod def classmethod(self): pass # Shadows builtin classmethod def meth(self) -> classmethod: pass # Stores 'classmethod', resolves to Foo.classmethod Foo.meth.annotations['return'] # <bound method Foo.classmethod of <class 'main.Foo'>> Sources confirm shadowing follows standard scope rules; future import changes storage/evaluation timing but not resolution order (8,25,28,30,31).
Citations:
- 1: Should a method name shadow outer names when quoted or with PEP563? python/mypy#18525
- 2: https://stackoverflow.com/questions/70596809/can-a-class-attribute-shadow-a-built-in-in-python
- 3: Class with a classmethod named
listfails when other methods uselistin annotations python/mypy#17517 - 4: Class __annotations__ has wrong values if class attr has the same name as it's provided type annotation python/cpython#98876
- 5: PEP 749 / 3.14 regression: Method annotation incorrectly resolves to method itself when method name matches imported type python/cpython#143306
- 6: https://peps.python.org/pep-0563/
- 7: https://peps.python.org/pep-0749/
Use builtins.list here; list[str] is resolving against ArtifactsAPI.list.
This class already defines list() at Line 249, so Line 2483 no longer refers to the builtin in this scope. That's why Ruff is raising A003, and it can break annotation evaluation at runtime and in type checkers. Use builtins.list[str] or rename the method.
Suggested fix
- notebook_ids: list[str],
+ notebook_ids: builtins.list[str],🧰 Tools
🪛 Ruff (0.15.9)
[error] 2483-2483: Python builtin is shadowed by method list from line 249
(A003)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/notebooklm/_artifacts.py` around lines 2483 - 2484, The parameter
annotation notebook_ids: list[str] is shadowing the class method list(), causing
A003; change the annotation to builtins.list[str] (and add an import for
builtins at the top if not present) so the type refers to the builtin list
rather than the class method; update any other annotations in this class that
use list[...] the same way (e.g., function/method signatures around generate_fn
and the class that defines list()).
| async def _generate_one(nb_id: str) -> None: | ||
| async with semaphore: | ||
| try: | ||
| status = await generate_fn(nb_id) | ||
| results[nb_id] = status | ||
| except Exception as e: # noqa: BLE001 | ||
| if error_mode == "fail_fast": | ||
| raise | ||
| results[nb_id] = e | ||
|
|
||
| await asyncio.gather( | ||
| *[_generate_one(nb_id) for nb_id in notebook_ids], | ||
| return_exceptions=error_mode == "collect_all", | ||
| ) | ||
|
|
||
| successful = sum(1 for r in results.values() if isinstance(r, GenerationStatus)) | ||
| logger.info( | ||
| "Bulk generate complete: %d/%d successful across %d notebooks", | ||
| successful, | ||
| total, | ||
| total, | ||
| ) |
There was a problem hiding this comment.
Route failed GenerationStatus values through the failure path.
Lines 2055-2070 already convert some RPC failures into GenerationStatus(status="failed", ...) instead of raising. generate_bulk() currently records those as successes, so error_mode="fail_fast" never trips on that path and the success log is inflated.
Suggested fix
async def _generate_one(nb_id: str) -> None:
async with semaphore:
try:
status = await generate_fn(nb_id)
+ if status.is_failed:
+ error = RuntimeError(
+ status.error or f"Generation failed for notebook {nb_id}"
+ )
+ if error_mode == "fail_fast":
+ raise error
+ results[nb_id] = error
+ return
results[nb_id] = status
except Exception as e: # noqa: BLE001
if error_mode == "fail_fast":
raise
results[nb_id] = e
@@
- successful = sum(1 for r in results.values() if isinstance(r, GenerationStatus))
+ successful = sum(
+ 1
+ for r in results.values()
+ if isinstance(r, GenerationStatus) and not r.is_failed
+ )🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/notebooklm/_artifacts.py` around lines 2527 - 2548, generate_bulk
currently treats returned GenerationStatus objects as successes; update
_generate_one to treat GenerationStatus(status="failed", ...) as a failure path:
after awaiting generate_fn(nb_id), if the result is a GenerationStatus with
status == "failed" then either raise (when error_mode == "fail_fast") or store
the GenerationStatus in results as an error (so return_exceptions/collect_all
sees it as failure). Refer to _generate_one, generate_fn, generate_bulk,
GenerationStatus and results in the diff and ensure the success counting and
logger.info use isinstance(r, GenerationStatus) and r.status == "succeeded" (or
inverse) so failed GenerationStatus values are not counted as successes.
| else: | ||
| conversation_history = self._build_conversation_history(conversation_id) |
There was a problem hiding this comment.
Add type narrowing assertion for consistency with ask().
The ask() method at line 104 includes assert conversation_id is not None for mypy type narrowing. This else-branch should do the same:
else:
+ assert conversation_id is not None # Type narrowing for mypy
conversation_history = self._build_conversation_history(conversation_id)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| else: | |
| conversation_history = self._build_conversation_history(conversation_id) | |
| else: | |
| assert conversation_id is not None # Type narrowing for mypy | |
| conversation_history = self._build_conversation_history(conversation_id) |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/notebooklm/_chat.py` around lines 233 - 234, The else-branch that calls
_build_conversation_history(conversation_id) lacks the same type-narrowing
assertion used in ask(), so add an assertion like assert conversation_id is not
None before calling self._build_conversation_history to satisfy mypy and ensure
conversation_id is treated as non-None in that branch (referencing the method
_build_conversation_history and the ask() pattern).
There was a problem hiding this comment.
🧹 Nitpick comments (1)
src/notebooklm/_chat.py (1)
39-39: Consider sorting__slots__alphabetically.Ruff RUF023 flags this as unsorted. While the current order follows a logical grouping (content → metadata → status), sorting alphabetically aids discoverability in larger classes.
♻️ Optional fix
- __slots__ = ("text", "is_answer", "reference", "conversation_id", "done") + __slots__ = ("conversation_id", "done", "is_answer", "reference", "text")🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/notebooklm/_chat.py` at line 39, The __slots__ tuple on the class in src/notebooklm/_chat.py is unsorted and triggers Ruff RUF023; update the __slots__ declaration (the "__slots__" symbol containing "text", "is_answer", "reference", "conversation_id", "done") to an alphabetically sorted tuple (e.g., "conversation_id", "done", "is_answer", "reference", "text") so the slot names are ordered lexicographically.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@src/notebooklm/_chat.py`:
- Line 39: The __slots__ tuple on the class in src/notebooklm/_chat.py is
unsorted and triggers Ruff RUF023; update the __slots__ declaration (the
"__slots__" symbol containing "text", "is_answer", "reference",
"conversation_id", "done") to an alphabetically sorted tuple (e.g.,
"conversation_id", "done", "is_answer", "reference", "text") so the slot names
are ordered lexicographically.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 568abf00-61a0-4d77-9325-2d1e9c52e431
📒 Files selected for processing (3)
src/notebooklm/_artifacts.pysrc/notebooklm/_chat.pysrc/notebooklm/_retry.py
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (4)
src/notebooklm/_chat.py (2)
257-263:⚠️ Potential issue | 🟠 MajorAdd explicit type narrowing before building follow-up history.
conversation_idis stillstr | Nonein this branch, but_build_conversation_history()expectsstr. Line 262 should mirror theask()narrowing pattern so mypy stays clean.Proposed fix
if is_new_conversation: conversation_id = str(uuid.uuid4()) conversation_history = None else: + assert conversation_id is not None # Type narrowing for mypy conversation_history = self._build_conversation_history(conversation_id)As per coding guidelines, "src/notebooklm/**/*.py: Use type hints and ensure type checking passes with mypy for
src/notebooklm."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/notebooklm/_chat.py` around lines 257 - 263, The branch that calls _build_conversation_history uses conversation_id which is still typed as str | None; narrow it to str before calling (mirror the ask() pattern) e.g., introduce a local narrowed variable or use an assert/type-check so mypy knows conversation_id is a str when passed into _build_conversation_history; update the is_new_conversation branch logic around the conversation_id variable and the call site to ensure _build_conversation_history always receives a plain str.
342-353:⚠️ Potential issue | 🟠 MajorReturn the resolved conversation ID in the final done chunk.
Line 352 currently returns only
server_conv_id. If the server does not emit one, consumers getNoneeven thoughfinal_conv_idexists and may already be cached.Proposed fix
# Final chunk with conversation_id - yield StreamChunk(conversation_id=server_conv_id, done=True) + yield StreamChunk(conversation_id=final_conv_id, done=True)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/notebooklm/_chat.py` around lines 342 - 353, The final yielded StreamChunk uses server_conv_id which can be None even when a valid final_conv_id (computed from server_conv_id or conversation_id) exists; update the final yield to return the resolved conversation id by passing final_conv_id (not server_conv_id) into StreamChunk(conversation_id=...) so consumers receive the cached/actual conversation id when server_conv_id is absent—change the yield after caching (where final_conv_id is computed) to StreamChunk(conversation_id=final_conv_id, done=True).src/notebooklm/_retry.py (2)
201-202:⚠️ Potential issue | 🟠 MajorReset
RetryManagerstate in__aenter__()to avoid stale reuse.Re-entering the same instance keeps prior
attempt,_current_delay, and_donevalues.Proposed fix
async def __aenter__(self) -> RetryManager: + self.attempt = 0 + self._current_delay = self.initial_delay + self._done = False return self🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/notebooklm/_retry.py` around lines 201 - 202, The RetryManager.__aenter__ currently returns self without resetting state; update __aenter__ in class RetryManager to reinitialize the retry fields (set self.attempt = 0, self._done = False and reset self._current_delay to the manager's initial delay value—e.g., self._initial_delay or self._base_delay whichever holds the configured start delay) so a reused instance doesn't carry over previous attempt/_current_delay/_done values; ensure the referenced initial-delay attribute exists or add one if necessary and keep __aenter__ idempotent.
121-128:⚠️ Potential issue | 🟠 MajorBackoff timing is off by one, and jitter can exceed
max_delay.All three implementations multiply
delaybefore sleep, so the first retry waitsinitial_delay * backoff_factor. Also, jitter is applied after clamp, so sleep can overshootmax_delay.Proposed fix pattern (apply in all three blocks)
- if isinstance(e, RateLimitError) and e.retry_after is not None: - delay = float(e.retry_after) + sleep_for = min(delay, max_delay) + if isinstance(e, RateLimitError) and e.retry_after is not None: + sleep_for = min(float(e.retry_after), max_delay) + delay = sleep_for logger.warning( "Rate limited — using server-suggested retry-after " "of %.1fs (attempt %d/%d)", - delay, + sleep_for, attempt, max_attempts, ) elif attempt < max_attempts: - if jitter: - delay = min(delay * backoff_factor, max_delay) - delay *= 0.75 + random.random() * 0.5 - else: - delay = min(delay * backoff_factor, max_delay) + if jitter: + sleep_for = min(delay * (0.75 + random.random() * 0.5), max_delay) + delay = min(delay * backoff_factor, max_delay) if attempt < max_attempts: logger.warning( "Transient error (%s) on attempt %d/%d — " "retrying in %.1fs", type(e).__name__, attempt, max_attempts, - delay, + sleep_for, ) - await asyncio.sleep(delay) + await asyncio.sleep(sleep_for)Also applies to: 224-229, 309-314
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/notebooklm/_retry.py` around lines 121 - 128, The retry backoff logic currently multiplies delay before sleeping and applies jitter after clamping, causing the first retry to wait initial_delay * backoff_factor and allowing jitter to exceed max_delay; fix by sleeping using the current delay (apply jitter to the current delay if jitter is enabled), clamp the jittered sleep value to max_delay, then after sleeping compute the next delay as next_delay = min(delay * backoff_factor, max_delay) and assign it to delay for the next iteration; update the three occurrences that use variables attempt, max_attempts, delay, backoff_factor, max_delay, and jitter so the first retry uses initial_delay (possibly jittered but not exceeding max_delay) and all jittered sleeps are clamped.
🧹 Nitpick comments (1)
src/notebooklm/_artifacts.py (1)
2628-2628: Line exceeds 100-character limit.This line is approximately 110 characters. As per coding guidelines, the maximum line length is 100 characters.
Suggested fix
if name_counts[base] > 1: - out_path = str(Path(output_dir) / _safe_filename(art.title or art.id, ext, f"-{name_counts[base] - 1}")) + suffix = f"-{name_counts[base] - 1}" + filename = _safe_filename(art.title or art.id, ext, suffix) + out_path = str(Path(output_dir) / filename) else:🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/notebooklm/_artifacts.py` at line 2628, The long assignment to out_path exceeds the 100-char limit; split it into clearer subexpressions and assign intermediate values to variables before building the final path. For example, compute base_name = art.title or art.id, suffix = f"-{name_counts[base] - 1}", filename = _safe_filename(base_name, ext, suffix) and then set out_path = str(Path(output_dir) / filename) so you keep the same behavior in the out_path assignment (references: out_path, Path, _safe_filename, art.title, art.id, name_counts[base]).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/notebooklm/_chat.py`:
- Line 39: StreamChunk.__slots__ is not alphabetically sorted which triggers
Ruff RUF023; update the __slots__ tuple in the StreamChunk class so the slot
names are sorted lexicographically (e.g., reorder ("text", "is_answer",
"reference", "conversation_id", "done") to the alphabetically sorted order) to
satisfy Ruff's rule.
- Around line 330-334: The code currently appends refs to accumulated_refs but
still yields StreamChunk(reference=ref) for every ref (symbols: refs,
accumulated_refs, StreamChunk), so change the loop to only yield when the ref is
not already in accumulated_refs: check "if ref not in accumulated_refs", then
append (or add to a set for O(1) checks) and yield StreamChunk(reference=ref);
otherwise skip yielding to enforce deduplication of emitted chunks.
---
Duplicate comments:
In `@src/notebooklm/_chat.py`:
- Around line 257-263: The branch that calls _build_conversation_history uses
conversation_id which is still typed as str | None; narrow it to str before
calling (mirror the ask() pattern) e.g., introduce a local narrowed variable or
use an assert/type-check so mypy knows conversation_id is a str when passed into
_build_conversation_history; update the is_new_conversation branch logic around
the conversation_id variable and the call site to ensure
_build_conversation_history always receives a plain str.
- Around line 342-353: The final yielded StreamChunk uses server_conv_id which
can be None even when a valid final_conv_id (computed from server_conv_id or
conversation_id) exists; update the final yield to return the resolved
conversation id by passing final_conv_id (not server_conv_id) into
StreamChunk(conversation_id=...) so consumers receive the cached/actual
conversation id when server_conv_id is absent—change the yield after caching
(where final_conv_id is computed) to StreamChunk(conversation_id=final_conv_id,
done=True).
In `@src/notebooklm/_retry.py`:
- Around line 201-202: The RetryManager.__aenter__ currently returns self
without resetting state; update __aenter__ in class RetryManager to reinitialize
the retry fields (set self.attempt = 0, self._done = False and reset
self._current_delay to the manager's initial delay value—e.g.,
self._initial_delay or self._base_delay whichever holds the configured start
delay) so a reused instance doesn't carry over previous
attempt/_current_delay/_done values; ensure the referenced initial-delay
attribute exists or add one if necessary and keep __aenter__ idempotent.
- Around line 121-128: The retry backoff logic currently multiplies delay before
sleeping and applies jitter after clamping, causing the first retry to wait
initial_delay * backoff_factor and allowing jitter to exceed max_delay; fix by
sleeping using the current delay (apply jitter to the current delay if jitter is
enabled), clamp the jittered sleep value to max_delay, then after sleeping
compute the next delay as next_delay = min(delay * backoff_factor, max_delay)
and assign it to delay for the next iteration; update the three occurrences that
use variables attempt, max_attempts, delay, backoff_factor, max_delay, and
jitter so the first retry uses initial_delay (possibly jittered but not
exceeding max_delay) and all jittered sleeps are clamped.
---
Nitpick comments:
In `@src/notebooklm/_artifacts.py`:
- Line 2628: The long assignment to out_path exceeds the 100-char limit; split
it into clearer subexpressions and assign intermediate values to variables
before building the final path. For example, compute base_name = art.title or
art.id, suffix = f"-{name_counts[base] - 1}", filename =
_safe_filename(base_name, ext, suffix) and then set out_path =
str(Path(output_dir) / filename) so you keep the same behavior in the out_path
assignment (references: out_path, Path, _safe_filename, art.title, art.id,
name_counts[base]).
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: a2f12969-acb3-40f4-b2c1-31e3722a7672
📒 Files selected for processing (3)
src/notebooklm/_artifacts.pysrc/notebooklm/_chat.pysrc/notebooklm/_retry.py
| if refs: | ||
| for ref in refs: | ||
| if ref not in accumulated_refs: | ||
| accumulated_refs.append(ref) | ||
| yield StreamChunk(reference=ref) |
There was a problem hiding this comment.
Reference deduplication is tracked but not enforced on emitted chunks.
Line 334 yields every reference, including duplicates; the dedupe check only gates accumulation.
Proposed fix
if refs:
for ref in refs:
if ref not in accumulated_refs:
accumulated_refs.append(ref)
- yield StreamChunk(reference=ref)
+ yield StreamChunk(reference=ref)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if refs: | |
| for ref in refs: | |
| if ref not in accumulated_refs: | |
| accumulated_refs.append(ref) | |
| yield StreamChunk(reference=ref) | |
| if refs: | |
| for ref in refs: | |
| if ref not in accumulated_refs: | |
| accumulated_refs.append(ref) | |
| yield StreamChunk(reference=ref) |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/notebooklm/_chat.py` around lines 330 - 334, The code currently appends
refs to accumulated_refs but still yields StreamChunk(reference=ref) for every
ref (symbols: refs, accumulated_refs, StreamChunk), so change the loop to only
yield when the ref is not already in accumulated_refs: check "if ref not in
accumulated_refs", then append (or add to a set for O(1) checks) and yield
StreamChunk(reference=ref); otherwise skip yielding to enforce deduplication of
emitted chunks.
There was a problem hiding this comment.
Actionable comments posted: 3
♻️ Duplicate comments (5)
src/notebooklm/_retry.py (2)
222-229:⚠️ Potential issue | 🟡 MinorSame jitter-after-clamp issue as the decorator.
The jitter multiplier is applied after clamping at
max_delay, allowing_current_delayto exceed the configured maximum:else: - self._current_delay = min( - self._current_delay * self.backoff_factor, self.max_delay - ) - if self.jitter: - self._current_delay *= 0.75 + random.random() * 0.5 + self._current_delay *= self.backoff_factor + if self.jitter: + self._current_delay *= 0.75 + random.random() * 0.5 + self._current_delay = min(self._current_delay, self.max_delay)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/notebooklm/_retry.py` around lines 222 - 229, The jitter multiplier is applied after clamping which can push _current_delay above max_delay; fix by applying jitter before enforcing the max and then assign the clamped result to self._current_delay. Concretely, compute the backoff candidate from self._current_delay * self.backoff_factor (or use error.retry_after), apply the jitter if self.jitter to that candidate, then set self._current_delay = min(candidate, self.max_delay) so the max_delay is never exceeded (references: _current_delay, backoff_factor, max_delay, jitter).
130-136:⚠️ Potential issue | 🟡 MinorJitter applied after clamping can exceed
max_delay.The backoff logic sleeps first (correct), then computes the next delay. However, jitter is applied after clamping at
max_delay, so the computed delay for subsequent iterations can exceed the documented maximum by up to 25%.Example: if
delayreaches 30s (max_delay) and jitter multiplier is 1.25, the next iteration's delay becomes 37.5s.if attempt < max_attempts: await asyncio.sleep(delay) # Apply backoff AFTER sleeping for the NEXT iteration if not (isinstance(e, RateLimitError) and e.retry_after is not None): - delay = min(delay * backoff_factor, max_delay) - if jitter: - delay *= 0.75 + random.random() * 0.5 + delay *= backoff_factor + if jitter: + delay *= 0.75 + random.random() * 0.5 + delay = min(delay, max_delay)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/notebooklm/_retry.py` around lines 130 - 136, The backoff computation currently applies jitter after clamping which can push delay above max_delay; update the logic in the retry loop where delay, backoff_factor, max_delay, and jitter are used (and RateLimitError / e.retry_after are checked) to first compute the next candidate delay = delay * backoff_factor, then apply jitter to that candidate if jitter is enabled, and finally clamp the result with delay = min(candidate_with_jitter, max_delay) so the post-jitter value never exceeds max_delay while preserving the existing sleep-before-backoff behavior.src/notebooklm/_artifacts.py (3)
2483-2483:⚠️ Potential issue | 🔴 CriticalUse
builtins.list[str]fornotebook_ids.
ArtifactsAPIalready defineslist()at Line 249, solist[str]here resolves against that class member during class body evaluation. This still triggers the same A003 shadowing problem and can break annotation evaluation at import time. Usebuiltins.list[str], like the other methods in this class.Suggested fix
- notebook_ids: list[str], + notebook_ids: builtins.list[str],Read-only verification
#!/bin/bash set -euo pipefail sed -n '249,252p' src/notebooklm/_artifacts.py sed -n '2481,2484p' src/notebooklm/_artifacts.py python - <<'PY' src = """ class Demo: def list(self): ... def generate_bulk(self, notebook_ids: list[str]): ... """ try: exec(src, {}) except Exception as exc: print(f"{type(exc).__name__}: {exc}") else: print("No runtime failure under this Python version; still verify Ruff/mypy for shadowing.") PYAs per coding guidelines, "Run ruff check on source and test directories to detect linting issues before committing."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/notebooklm/_artifacts.py` at line 2483, The type annotation for the parameter notebook_ids in ArtifactsAPI (the method that currently uses notebook_ids: list[str]) shadows the class member list and can fail during class body evaluation; change that annotation to use builtins.list[str] (e.g., notebook_ids: builtins.list[str]) to avoid resolving against the ArtifactsAPI.list method—update the method signature where notebook_ids is declared and run ruff/mypy to verify the shadowing warning is gone.
2604-2631:⚠️ Potential issue | 🟠 MajorThe suffixing logic can still generate duplicate output paths.
This only deduplicates exact base-name matches. A real title like
Report-1still collides with the secondReport, because both resolve toReport-1.<ext>. That can schedule two concurrent downloads to the same file again.More robust fix
- def _safe_filename(title: str, ext: str, suffix: str = "") -> str: + def _safe_filename(title: str, artifact_id: str, ext: str) -> str: safe = "".join(c if c.isalnum() or c in " -_" else "_" for c in title) - return f"{safe[:80]}{suffix}.{ext}" + stem = safe[:60].rstrip(" ._-") or artifact_id + return f"{stem}-{artifact_id}.{ext}" @@ - name_counts: dict[str, int] = {} specs: list[tuple[str, str]] = [] for art in completed: - base = _safe_filename(art.title or art.id, ext) - name_counts[base] = name_counts.get(base, 0) + 1 - if name_counts[base] > 1: - out_path = str(Path(output_dir) / _safe_filename(art.title or art.id, ext, f"-{name_counts[base] - 1}")) - else: - out_path = str(Path(output_dir) / base) + out_path = str( + Path(output_dir) / _safe_filename(art.title or art.id, art.id, ext) + ) specs.append((art.id, out_path))🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/notebooklm/_artifacts.py` around lines 2604 - 2631, The current suffixing in _safe_filename/name_counts can still produce duplicate out_paths (e.g., a title "Report-1" collides with auto-suffixed "Report-1"); change the loop that builds specs so that for each art you generate a candidate filename and then increment a numeric suffix until the fully qualified out_path (Path(output_dir)/candidate) is unique, using a set like used_paths to track already chosen out_paths; keep using _safe_filename to sanitize titles and preserve the extension logic around extensions/get, but replace the simple name_counts==>out_path branch with a while-loop that checks membership in used_paths and only adds the final unique out_path to specs and to used_paths (reference symbols: _safe_filename, name_counts, specs, completed).
2530-2545:⚠️ Potential issue | 🟠 MajorTreat failed
GenerationStatusvalues as failures.
_call_generate()already returnsGenerationStatus(status="failed", ...)for some RPC failures at Lines 2055-2070. This path is still recorded as a success here, soerror_mode="fail_fast"never trips on it and Line 2545 overcounts successful generations.Possible fix
async def _generate_one(nb_id: str) -> None: async with semaphore: try: status = await generate_fn(nb_id) + if status.is_failed: + error = RuntimeError( + status.error or f"Generation failed for notebook {nb_id}" + ) + if error_mode == "fail_fast": + raise error + results[nb_id] = error + return results[nb_id] = status except Exception as e: # noqa: BLE001 if error_mode == "fail_fast": raise results[nb_id] = e @@ - successful = sum(1 for r in results.values() if isinstance(r, GenerationStatus)) + successful = sum( + 1 + for r in results.values() + if isinstance(r, GenerationStatus) and not r.is_failed + )🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/notebooklm/_artifacts.py` around lines 2530 - 2545, The code currently treats GenerationStatus objects as successes even when they represent failures, so update _generate_one to detect a returned GenerationStatus with status == "failed": after calling status = await generate_fn(nb_id), if isinstance(status, GenerationStatus) and status.status == "failed" then if error_mode == "fail_fast" raise an exception (e.g., RuntimeError with status details) else store an Exception-like value in results[nb_id] (or the GenerationStatus itself but ensure callers treat it as failure); also change the successful tally to count only GenerationStatus instances whose .status != "failed" (i.e., successful = sum(1 for r in results.values() if isinstance(r, GenerationStatus) and r.status != "failed")). Use the existing symbols _generate_one, generate_fn, results, error_mode, and GenerationStatus to locate and implement these checks.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/notebooklm/_artifacts.py`:
- Around line 2540-2543: The current asyncio.gather call schedules every
_generate_one upfront and in fail_fast mode (return_exceptions=False) an
exception will be raised but sibling coroutines are not cancelled; modify the
code so you create tasks via asyncio.create_task for each notebook (e.g., tasks
= [asyncio.create_task(_generate_one(nb_id)) for nb_id in notebook_ids]) and
then: if error_mode == "collect_all" await asyncio.gather(*tasks,
return_exceptions=True); else try to await asyncio.gather(*tasks) and in the
except block cancel all pending tasks (for t in tasks: if not t.done():
t.cancel()), then await asyncio.gather(*tasks, return_exceptions=True) to drain
cancellations and re-raise the original exception so generate_fn stops other
notebook processing. Ensure you reference _generate_one and the asyncio.gather
invocation when updating the code.
In `@src/notebooklm/_chat.py`:
- Around line 28-53: The StreamChunk class (from _chat) is part of the public
ask_stream() API but isn’t exported from the package root; import StreamChunk
into __init__.py (e.g., from ._chat import StreamChunk) and add "StreamChunk" to
the module’s __all__ list so users can do from notebooklm import StreamChunk and
IDE/type hints work correctly.
In `@src/notebooklm/_retry.py`:
- Around line 316-323: The jitter is applied after clamping which can exceed
max_delay; in the retry loop (the block using variables delay, backoff_factor,
max_delay, jitter and exception type RateLimitError) compute the
increased/backoff delay first, apply jitter to that value if jitter is set, and
then clamp with delay = min(new_delay, max_delay); keep the RateLimitError
branch setting delay = float(e.retry_after) unchanged. This ensures jitter is
included before applying the max_delay cap in the retry logic.
---
Duplicate comments:
In `@src/notebooklm/_artifacts.py`:
- Line 2483: The type annotation for the parameter notebook_ids in ArtifactsAPI
(the method that currently uses notebook_ids: list[str]) shadows the class
member list and can fail during class body evaluation; change that annotation to
use builtins.list[str] (e.g., notebook_ids: builtins.list[str]) to avoid
resolving against the ArtifactsAPI.list method—update the method signature where
notebook_ids is declared and run ruff/mypy to verify the shadowing warning is
gone.
- Around line 2604-2631: The current suffixing in _safe_filename/name_counts can
still produce duplicate out_paths (e.g., a title "Report-1" collides with
auto-suffixed "Report-1"); change the loop that builds specs so that for each
art you generate a candidate filename and then increment a numeric suffix until
the fully qualified out_path (Path(output_dir)/candidate) is unique, using a set
like used_paths to track already chosen out_paths; keep using _safe_filename to
sanitize titles and preserve the extension logic around extensions/get, but
replace the simple name_counts==>out_path branch with a while-loop that checks
membership in used_paths and only adds the final unique out_path to specs and to
used_paths (reference symbols: _safe_filename, name_counts, specs, completed).
- Around line 2530-2545: The code currently treats GenerationStatus objects as
successes even when they represent failures, so update _generate_one to detect a
returned GenerationStatus with status == "failed": after calling status = await
generate_fn(nb_id), if isinstance(status, GenerationStatus) and status.status ==
"failed" then if error_mode == "fail_fast" raise an exception (e.g.,
RuntimeError with status details) else store an Exception-like value in
results[nb_id] (or the GenerationStatus itself but ensure callers treat it as
failure); also change the successful tally to count only GenerationStatus
instances whose .status != "failed" (i.e., successful = sum(1 for r in
results.values() if isinstance(r, GenerationStatus) and r.status != "failed")).
Use the existing symbols _generate_one, generate_fn, results, error_mode, and
GenerationStatus to locate and implement these checks.
In `@src/notebooklm/_retry.py`:
- Around line 222-229: The jitter multiplier is applied after clamping which can
push _current_delay above max_delay; fix by applying jitter before enforcing the
max and then assign the clamped result to self._current_delay. Concretely,
compute the backoff candidate from self._current_delay * self.backoff_factor (or
use error.retry_after), apply the jitter if self.jitter to that candidate, then
set self._current_delay = min(candidate, self.max_delay) so the max_delay is
never exceeded (references: _current_delay, backoff_factor, max_delay, jitter).
- Around line 130-136: The backoff computation currently applies jitter after
clamping which can push delay above max_delay; update the logic in the retry
loop where delay, backoff_factor, max_delay, and jitter are used (and
RateLimitError / e.retry_after are checked) to first compute the next candidate
delay = delay * backoff_factor, then apply jitter to that candidate if jitter is
enabled, and finally clamp the result with delay = min(candidate_with_jitter,
max_delay) so the post-jitter value never exceeds max_delay while preserving the
existing sleep-before-backoff behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 34be8b81-71b8-471e-b17f-5d5a51820acf
📒 Files selected for processing (3)
src/notebooklm/_artifacts.pysrc/notebooklm/_chat.pysrc/notebooklm/_retry.py
- Add _retry module: @retry_on_transient_errors decorator, retry_with_backoff() helper, and RetryManager context manager with exponential backoff and jitter - Add ask_stream() to ChatAPI: async generator yielding text chunks and references in real time as the response generates - Add generate_bulk() and download_all() to ArtifactsAPI: run artifact generation/downloads across multiple notebooks concurrently with concurrency limits and per-item error collection
There was a problem hiding this comment.
♻️ Duplicate comments (2)
src/notebooklm/_artifacts.py (1)
2530-2558:⚠️ Potential issue | 🟠 MajorFailed
GenerationStatusvalues are counted as successes.When
generate_fnreturns aGenerationStatuswithstatus="failed"(e.g., from rate limit handling at lines 2063-2070), the current implementation stores it as a success and counts it in the success tally. This inflates the success count and preventserror_mode="fail_fast"from triggering on these soft failures.From the relevant code snippet,
GenerationStatus.is_failedreturnsTruewhenstatus == "failed".Suggested fix
async def _generate_one(nb_id: str) -> None: async with semaphore: try: status = await generate_fn(nb_id) + if status.is_failed: + error = RuntimeError( + status.error or f"Generation failed for notebook {nb_id}" + ) + if error_mode == "fail_fast": + raise error + results[nb_id] = error + return results[nb_id] = status except Exception as e: # noqa: BLE001 if error_mode == "fail_fast": raise results[nb_id] = eAnd update the success count:
- successful = sum(1 for r in results.values() if isinstance(r, GenerationStatus)) + successful = sum( + 1 for r in results.values() + if isinstance(r, GenerationStatus) and not r.is_failed + )🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/notebooklm/_artifacts.py` around lines 2530 - 2558, The code treats any returned GenerationStatus as a success; fix _generate_one and the final success tally so GenerationStatus objects with status=="failed" are considered failures: after status = await generate_fn(nb_id) in _generate_one check if isinstance(status, GenerationStatus) and status.is_failed; if so, in fail_fast mode raise an exception (or raise a RuntimeError wrapping the failed GenerationStatus) so the outer exception handling cancels tasks, otherwise store the failed GenerationStatus in results (or store an Exception wrapper) but do not treat it as successful; also update the final success count to only count entries where isinstance(r, GenerationStatus) and not r.is_failed.src/notebooklm/_retry.py (1)
226-230:⚠️ Potential issue | 🟡 MinorJitter can push delay beyond
max_delay.Lines 226-227 clamp the delay to
max_delay, but lines 229-230 then multiply by the jitter factor (0.75-1.25). With a max multiplier of 1.25, a clamped delay of 30.0s becomes 37.5s, violating the documented contract.Suggested fix
else: - self._current_delay = min( - self._current_delay * self.backoff_factor, self.max_delay - ) + self._current_delay *= self.backoff_factor if self.jitter: self._current_delay *= 0.75 + random.random() * 0.5 + self._current_delay = min(self._current_delay, self.max_delay)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/notebooklm/_retry.py` around lines 226 - 230, The current update to self._current_delay multiplies by backoff_factor then applies jitter, which can push the value above self.max_delay; change the logic in the backoff step of the retry iterator (where _current_delay, backoff_factor, max_delay, and jitter are used) so that jitter is applied before the final clamp or ensure you re-clamp after applying jitter — e.g., compute the increased delay, apply jitter if enabled, then set self._current_delay = min(computed_delay, self.max_delay) to guarantee the delay never exceeds max_delay.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Duplicate comments:
In `@src/notebooklm/_artifacts.py`:
- Around line 2530-2558: The code treats any returned GenerationStatus as a
success; fix _generate_one and the final success tally so GenerationStatus
objects with status=="failed" are considered failures: after status = await
generate_fn(nb_id) in _generate_one check if isinstance(status,
GenerationStatus) and status.is_failed; if so, in fail_fast mode raise an
exception (or raise a RuntimeError wrapping the failed GenerationStatus) so the
outer exception handling cancels tasks, otherwise store the failed
GenerationStatus in results (or store an Exception wrapper) but do not treat it
as successful; also update the final success count to only count entries where
isinstance(r, GenerationStatus) and not r.is_failed.
In `@src/notebooklm/_retry.py`:
- Around line 226-230: The current update to self._current_delay multiplies by
backoff_factor then applies jitter, which can push the value above
self.max_delay; change the logic in the backoff step of the retry iterator
(where _current_delay, backoff_factor, max_delay, and jitter are used) so that
jitter is applied before the final clamp or ensure you re-clamp after applying
jitter — e.g., compute the increased delay, apply jitter if enabled, then set
self._current_delay = min(computed_delay, self.max_delay) to guarantee the delay
never exceeds max_delay.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: ad76c630-c727-45ed-8d0e-5c0c4876c013
📒 Files selected for processing (4)
src/notebooklm/__init__.pysrc/notebooklm/_artifacts.pysrc/notebooklm/_chat.pysrc/notebooklm/_retry.py
✅ Files skipped from review due to trivial changes (1)
- src/notebooklm/_chat.py
|
@aayushprsingh thanks for the PR. This PR covers three different features. It is usually a good idea to separate them into 3 PRs instead of one. |
2 participants
Summary
Adds three new capabilities to the library:
1. Retry utilities (
_retry.py)Exponential backoff with jitter for transient API errors. Three APIs:
@retry_on_transient_errors()decoratorretry_with_backoff(coro)one-linerRetryManagercontext manager for fine-grained controlRespects
RateLimitError.retry_afterand skips auth errors.2. Streaming chat (
ask_streamin_chat.py)ChatAPI.ask_stream()is an async generator that yields text chunks and references in real time as the response generates — no waiting for the full answer.3. Bulk artifact operations (
generate_bulk+download_allin_artifacts.py)Run artifact generation or downloads across multiple notebooks concurrently with concurrency limits and per-item error collection.
Testing
ruff --ignore E501with zero errorsSummary by CodeRabbit