Something went wrong. Try again.
Fetches every record from a repo collection using cursor-based pagination. listRecords returns at most 100 per call — this loops until the cursor is exhausted. Works unauthenticated for public repos; swap the base URL for a PDS endpoint to read private collections.
Something went wrong. Try again.
891 B · 34 lines · Created 1mo ago
Python
12345678910111213141516171819202122232425262728293031323334import httpx
def list_all_records(repo: str, collection: str, pds: str = "https://bsky.social") -> list: records = [] cursor = None
while True: params = {"repo": repo, "collection": collection, "limit": 100} if cursor: params["cursor"] = cursor
resp = httpx.get(f"{pds}/xrpc/com.atproto.repo.listRecords", params=params) resp.raise_for_status() data = resp.json()
batch = data.get("records", []) records.extend(batch)
cursor = data.get("cursor") if not cursor or not batch: break
return records
if __name__ == "__main__": import json
posts = list_all_records( repo="distributed.camp", collection="app.bsky.feed.post", ) print(f"{len(posts)} posts") print(json.dumps(posts[0], indent=2))