DEV Community: Susanth P

Turning Reddit threads into RAG-ready Markdown

Susanth P — Thu, 22 Oct 2026 18:40:49 +0000

AI assistance disclosure: this tutorial was drafted with AI assistance.

A collection of thread text becomes more useful for retrieval when each chunk retains its source, timestamp, and position in the conversation. A flat string loses the distinction between a question and its replies. This tutorial starts with an existing JSON export that you are entitled to process. It does not collect data from Reddit or require a particular paid service.

Keep provenance alongside the text

Use a small normalized record: id, parentId, threadId, title, text, permalink, and createdAt. Omit usernames unless the application has a specific need for them. Store a collection timestamp separately from the original publication timestamp. These are different facts and matter when a source changes.

The example fixture is synthetic. It demonstrates formatting, not an actual Reddit discussion:

[
  {"id":"post-example","parentId":null,"threadId":"post-example","title":"How should a team review alerts?","text":"We want fewer duplicate notifications.","permalink":"https://example.com/thread","createdAt":"2026-10-22T09:00:00Z"},
  {"id":"reply-example","parentId":"post-example","threadId":"post-example","title":"","text":"Keep a stable event ID at the destination.","permalink":"https://example.com/thread#reply","createdAt":"2026-10-22T09:05:00Z"}
]

Make bounded, self-contained chunks

This standard-library Python example repeats the title and source metadata on each chunk. The limit is in characters, not model tokens. A production pipeline should use the tokenizer and limits of its chosen embedding model.

import hashlib
import json
import re

FENCE = chr(96) * 3

def plain(value):
    # Exported content is untrusted. Prevent it from closing our fence.
    return str(value or "").replace(FENCE, "~~~").replace("\r\n", "\n")

def split_text(text, limit=1200):
    if limit < 1:
        raise ValueError("limit must be positive")
    text = plain(text).strip()
    while text:
        cut = min(limit, len(text))
        if cut < len(text):
            boundary = text.rfind(" ", 0, cut)
            if boundary > limit // 2:
                cut = boundary
        yield text[:cut].strip()
        text = text[cut:].lstrip()

def chunks(records):
    titles = {r["id"]: r.get("title", "") for r in records}
    for record in records:
        body = record.get("text", "")
        title = titles.get(record.get("threadId"), "") or record.get("title", "")
        for index, part in enumerate(split_text(body)):
            # Content hash changes when the normalized chunk changes.
            identity = f'{record["id"]}:{index}'
            digest = hashlib.sha256(part.encode("utf-8")).hexdigest()
            metadata = {k: record.get(k) for k in
                        ("id", "threadId", "parentId", "permalink", "createdAt")}
            metadata.update(chunkId=identity, contentHash=digest, part=index)
            document = (
                f'# {plain(title)}\n\n'
                f'{FENCE}json\n{json.dumps(metadata, ensure_ascii=False)}\n{FENCE}\n\n'
                f'## Quoted source content\n\n{FENCE}text\n{part}\n{FENCE}\n'
            )
            yield {"id": identity, "metadata": metadata, "markdown": document}

if __name__ == "__main__":
    with open("threads.json", encoding="utf-8") as stream:
        records = json.load(stream)
    with open("chunks.jsonl", "w", encoding="utf-8") as stream:
        for item in chunks(records):
            stream.write(json.dumps(item, ensure_ascii=False) + "\n")

The fenced source section is a presentation convention, not a security boundary. When these documents reach an LLM, treat their contents as untrusted data. A source might contain instructions that conflict with your application. Keep system instructions outside the retrieved text and do not give retrieved content authority over tool use.

Handle updates and deletions deliberately

Use chunkId as an upsert key and contentHash to avoid unnecessary embedding work. If a shorter revision creates fewer chunks, remove stale trailing chunks for that record. A missing item in a partial export is not proof of deletion; reconcile only against an authoritative status or a defined retention policy. Keep an inventory of emitted chunk IDs for each source record.

Store parentId as metadata so retrieval can fetch nearby context. A reply such as “that worked” is not meaningful without its parent. Avoid concatenating an entire long thread into one document merely to preserve order. Retrieve a bounded parent or sibling window after finding an initial match.

Evaluate retrieval before adding more data

Write a few questions with known supporting passages. Check whether retrieval returns the right passage and whether the answer links to its source. Include a question the export cannot answer, so abstention is tested too. More chunks do not automatically produce better answers.

This script prepares Markdown and JSONL only. It does not create embeddings, validate all possible export schemas, or provide deletion synchronization. Those should be explicit later steps, selected for the actual application and its data permissions.

Handling Google Trends 429 errors: retries, caching and data checks

Susanth P — Wed, 21 Oct 2026 16:03:26 +0000

A Google Trends job can fail in two different ways: it can stop returning data, or it can return data that your analysis interprets incorrectly. Treating both as a retry problem makes a pipeline harder to debug.

This tutorial separates request control from data validation. The Python example uses a simulated reader, so you can experiment without sending requests to Google.

AI disclosure: This article and its code examples were generated by an AI coding agent at the author's request. The retry example was checked with offline tests. It illustrates a control-flow pattern, not a measured Google rate limit or a guarantee of access.

1. Reduce the work before retrying it

An HTTP 429 response means the server considers the request rate too high. It does not identify a universal safe request interval. RFC 6585, section 4 also allows the server to send a Retry-After header.

For a Trends workflow, start with the smallest report your analysis needs. A timeline, regional breakdown and related-query report are separate pieces of work. Avoid requesting every panel simply because a client exposes it.

Make repeated jobs reuse successful results. A useful cache identity includes:

The ordered search-term or topic list.
Geography, category and search property.
The requested date range and report type.
Any parameters that change the scale or resolution.

Resolve a rolling period such as “the past twelve months” to explicit dates in your job metadata. Give cached results an appropriate expiry: a fixed historical range can still be revised or sampled differently, while a rolling range also changes its boundaries.

Keep failed requests separate from cached successful reports. An unavailable panel should not overwrite yesterday's valid result with an empty list.

2. Give retries an explicit stopping point

This small helper retries only a designated rate-limit exception. An adapter around your chosen client would translate its actual exception and, if available, a parsed server delay into RateLimited.

import random
import time


class RateLimited(Exception):
    def __init__(self, retry_after_seconds=None):
        super().__init__("The upstream reader was rate limited")
        self.retry_after_seconds = retry_after_seconds


def retry_read(read_once, *, attempts=3, wait_budget=60,
               sleep=time.sleep, jitter=random.random):
    if attempts < 1 or wait_budget < 0:
        raise ValueError("Invalid retry limits")

    waited = 0.0
    for attempt in range(attempts):
        try:
            return read_once()
        except RateLimited as exc:
            if attempt == attempts - 1:
                raise

            backoff = min(30, 2 ** attempt) + jitter()
            server_delay = max(0, exc.retry_after_seconds or 0)
            delay = max(backoff, server_delay)
            if waited + delay > wait_budget:
                raise  # Defer the job instead of retrying too early.

            sleep(delay)
            waited += delay

attempts=3 means three calls in total, including the first one. wait_budget limits accumulated planned sleep, not total execution time; each underlying request also needs a timeout. The numbers here are illustrative policy choices.

The helper does not parse HTTP headers. In an HTTP adapter, Retry-After can contain delay seconds or an HTTP date; parse it before constructing the exception. If the server requests more waiting than the job allows, defer or stop the job. Do not shorten that delay to squeeze in another attempt.

Use one layer to own retries. Wrapping a client that already retries internally can multiply the number of upstream calls. Similarly, per-worker backoff does not control the combined request rate of ten workers; coordinate concurrency and scheduling across the job.

3. Test the failure path without a live endpoint

Add this below the helper and run the file with Python:

outcomes = iter([RateLimited(5), {"values": [12, 18, 15]}])
waits = []


def fake_read():
    outcome = next(outcomes)
    if isinstance(outcome, Exception):
        raise outcome
    return outcome


result = retry_read(fake_read, sleep=waits.append, jitter=lambda: 0)
assert waits == [5]
assert result == {"values": [12, 18, 15]}
print("Offline retry check passed")

The injected sleep function records the delay instead of waiting. Also test permanent rate limiting, a server delay above the wait budget and unrelated exceptions. Unrelated exceptions should propagate immediately, rather than being disguised as rate limiting.

4. Validate the report before using it

A successful network request is only the first check. Preserve the request parameters and collection time with the result, then verify the panels your analysis actually requires.

Keep unavailable, empty and complete results distinct.
Exclude a partial current time bucket when comparing complete periods.
Compare terms using a shared comparison scale. Two separately normalized reports that each peak at 100 do not establish equal search volume.
Preserve missing values rather than silently treating them as observed zeroes.

Google's explanation of Trends data describes sampled, normalized search interest. A value of 100 is an index on the report's scale, not 100 searches. A regional index is not a count of users, and selected search terms do not measure product market share.

5. Record enough context to diagnose the next failure

Log the report identity, attempt count, response status, chosen delay and completeness outcome. Avoid putting credentials or session cookies in those logs. Separate “the server refused the request” from “the report was valid but had no usable observations.”

If you maintain an older workflow, note that the original pytrends repository was archived on 17 April 2025. Pinning dependencies preserves an environment; it does not freeze an upstream service's behavior.

Reduce duplicate work, bound the retry path and retain the meaning of the data alongside the numbers. Those controls make a pipeline easier to inspect even when an upstream service remains unavailable.