Code-smell quick wins: dedupe streaming helpers, dead-code cleanup, project retry - #20
Merged
Merged
Conversation
The three streaming migrators (logs, datasets, experiments) each carried byte-for-byte copies of _is_http_413, _approx_event_size_bytes, _count_attachment_refs, the HTTP 413 constant, and a near-identical _dump_oversize_event_summary method (differing only by output dir, filename prefix, dest-id key name, and log wording). Move these to single shared functions in streaming_utils (is_http_413, approx_event_size_bytes, count_attachment_refs, dump_oversize_event_summary) and have all three migrators call them. The dump function takes the per-resource out_dir / filename_prefix / event_label / dest_id_field so behavior (including the exact summary schema and log messages) is preserved. Drops the now-unused httpx import from datasets and experiments. Behavior-preserving refactor: all existing tests pass unchanged. Adds test_streaming_helpers.py to pin the shared helpers directly (previously only covered indirectly through the migration paths). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
- Remove _get_client_resource_attr and _handle_api_response_to_list from ResourceMigrator; both had zero callers anywhere in the package. - record_failure stored failed_at as str(Path(__file__).stat().st_mtime) — base.py's own mtime, a static value identical for every failure. Replace with datetime.now(UTC).isoformat() so the recorded time is real. No test referenced these; full suite passes unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
list_projects (per-page) and create_project called raw_request directly, bypassing the adaptive retry/backoff that every other API call in the tool uses. Wrap both in with_retry so transient 429/5xx during project discovery and creation are retried consistently. Happy-path behavior is unchanged (with_retry calls the request once on success); only transient-failure behavior improves. Project methods are mocked wholesale in tests, so the suite passes unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The three streaming migrators each re-read byte-batching config with an identical try/except, and each computed self._insert_max_bytes from insert_max_request_bytes * insert_request_headroom_ratio — a value that was never read anywhere (dead), and whose fallback default (0.5) disagreed with the canonical MigrationConfig default (0.75). - Delete the dead _insert_max_bytes computation from all three migrators (removes the drift entirely rather than "fixing" an unused default). - Add StreamingConfig.resolve(source, dest) in streaming_utils as the single place that resolves the flush/fetch knobs, with module-constant defaults (STREAMING_FLUSH_MAX_ROWS/BYTES, STREAMING_MAX_EVENT_BYTES, STREAMING_EVENT_FETCH_GROUP_SIZE). The migrator ClassVars now alias those constants. Test mechanism change (not a product-behavior change): the three resume-after-insert-failure tests forced per-row flushing by monkeypatching the SDK_FLUSH_MAX_ROWS ClassVar; they now set events_flush_max_rows=1 on a MigrationConfig — the actual user-facing lever — which is both cleaner and lets resolve use plain module-constant defaults. Full suite: 293 passing. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…res (R8) _get_source_user_email, _find_dest_user_id_by_email, and _invite_user_to_dest_org were duplicated near-verbatim across the ACL and group migrators (only the with_retry op-name strings differed), and each swallowed every exception and cached None permanently — so a transient auth/network/5xx failure became a permanent "no such user". Extract a single UserResolver (braintrust_migrate/user_resolver.py) used by both migrators. with_retry re-raises the original exception, so a 404 propagates as httpx.HTTPStatusError(404): the resolver now caches the negative result only on a genuine 404, and on any other failure logs a warning and does NOT cache, leaving a later attempt able to succeed. The duplicated methods and their three per-migrator caches are removed; both migrators delegate to the resolver. Op-name suffix keeps telemetry distinguishable per caller. Unit suite (293) and the ACL/group user-mapping integration flow tests pass; the 404-vs-transient change only affects the error path (happy path is unchanged). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
streaming_pipeline (env MIGRATION_STREAMING_PIPELINE) was defined, parsed,
and constructed but never read anywhere in the product code — the streaming
loop is strictly sequential (fetch page, insert, fetch next). The README,
however, documented it as a working, default-on "pipelined event streaming"
feature ("prefetch the next BTQL page while inserting the current batch").
Remove the dead config field/env-parse/constructor arg, and correct the
README (env table, the Pipelined Event Streaming section, the parallelization
diagram, tuning tips, and example .env blocks) so it no longer claims a
feature that doesn't exist. Drop the streaming_pipeline assertions from the
unit config tests and the field from the live e2e scenario matrix (the two
scenarios remain distinct via their concurrency settings; renamed to
concurrent/sequential).
No runtime behavior change — the knob never affected anything. Implementing
real pipelined prefetch remains possible future work.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The dataset and experiment migrators each defined four progress-hook lambdas (on_fetch/on_page/on_insert/on_done) that built byte-identical ~95-line dicts, differing only by the resource label and the source/dest id-field keys. Add build_stream_progress(phase, info, state, *, resource, id_fields) and make_stream_progress_hooks(...) in streaming_utils, and have both migrators pass a single descriptor instead of the four hand-written dicts. The emitted payloads are unchanged (per-phase sourcing preserved: fetch/page/done pull totals from the loop info; insert pulls committed totals from state; cursor truncation and the resource/id-field keys are identical), so the CLI progress display contract is preserved. Scope: datasets + experiments (which share the buffered loop). Logs keeps its own progress functions for now — its payload shape differs and it folds in with R1 (logs onto the shared loop). Adds build_stream_progress / make_stream_progress_hooks unit tests. Suite: 294. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Doug Guthrie (dpguthrie)
force-pushed
the
chore/streaming-quick-wins
branch
from
June 8, 2026 20:32
836c668 to
7df4e75
Compare
- datasets.py: `_on_single_413` referenced an unbound `source_dataset_id`
(only a comprehension/loop variable elsewhere), so a single dataset event
hitting HTTP 413 raised NameError — aborting the migration and skipping the
oversize diagnostic instead of writing it and propagating the 413. Bind
`source_dataset_id = event.get("dataset_id")` in the closure, matching the
experiments migrator. (Latent bug present since before this branch; fixed
here since the refactor touches this code.)
- streaming_utils.py: drop the vestigial `_p=progress` default-arg in
make_stream_progress_hooks' lambda — it's the loop-capture idiom, but there
is no loop and `progress` is a stable captured param.
Adds test_dataset_oversize_413.py, which drives a single-event 413 through the
dataset streaming path (fails with NameError without the fix).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
First, low-risk batch from the code-smell audit. Strictly behavior-preserving (one bug fix called out below) — no existing test was modified; the suite went from 288 → 293 (only additions).
Changes
R2 + R3 — Hoist duplicated streaming helpers (
braintrust_migrate/streaming_utils.py)The three streaming migrators each carried byte-for-byte copies of
_is_http_413,_approx_event_size_bytes,_count_attachment_refs, the413constant, and a near-identical_dump_oversize_event_summary(differing only by output dir, filename prefix, dest-id key, and log wording). Moved to single shared functions; each migrator now calls them. The dump function takes per-resourceout_dir/filename_prefix/event_label/dest_id_fieldso the exact summary schema and log messages are preserved. Dropped the now-unusedhttpximport from datasets/experiments. Net −19 lines.R12 — Dead code + bogus timestamp (
resources/base.py)_get_client_resource_attrand_handle_api_response_to_list(zero callers anywhere).record_failurestoredfailed_at = str(Path(__file__).stat().st_mtime)— base.py's own mtime, a static value identical for every failure. Nowdatetime.now(UTC).isoformat().R11 — Route project list/create through
with_retry(client.py)list_projects(per page) andcreate_projectcalledraw_requestdirectly, bypassing the adaptive retry every other call uses. Wrapped both. Happy-path identical; only transient-failure behavior improves.Test deviation report (what the audit asked for)
tests/unit/test_streaming_helpers.py(5 tests) to pin the newly-shared helpers, which were previously only covered indirectly through the migration paths.failed_at) was uncovered by any test and is a clear fix.Deliberately deferred
run_dry_run/validateconsolidation) — more CLI-test-sensitive; separate.stream_btql_sorted_events_buffered, R4 shared base, R1 logs onto the shared loop) are the high-value/medium-risk core and will need test changes (monkeypatched writer names,EventsStreamStateJSON keys, progress payloads) — tracked for a follow-up.🤖 Generated with Claude Code