FlashData Python SDK

English | 简体中文

FlashData · API documentation

Google search results and YouTube data for Python applications, research pipelines, search monitoring, and retrieval workflows.

Get web, news, shopping, image, video, and local search results; discover YouTube videos, read metadata and existing caption tracks; submit background jobs; and process video ID files in bulk. The SDK includes synchronous and asynchronous clients, typed parameters, local input validation, and structured errors. Python 3.10 or newer is required.

What is FlashData?

FlashData provides APIs for Google search results, YouTube data, and cloud media delivery. Developers and data teams use these APIs for search monitoring, research, analytics, content discovery, and retrieval workflows for AI applications.

Retrieve web, news, shopping, image, video, and local search results; discover YouTube videos and read metadata, existing caption tracks, channel data, and license labels. Use realtime requests, background jobs, mixed batches, or bulk video ID files to fit your workload. Media downloads deliver audio or video into your own cloud storage. See the API documentation and pricing.

Install

pip install flashdata

This command requires FlashData's official PyPI publication. For local testing before release, use installation from source.

Configure your API key

Obtain an API key from your FlashData account and pass it through the client's api_key argument. The examples below use os.getenv("FLASHDATA_API_KEY") to read the key from your environment. You can also pass a key supplied by your secret manager.

The SDK does not automatically load .env files. Load application configuration before calling os.getenv if you use them.

# macOS / Linux
export FLASHDATA_API_KEY="YOUR_API_KEY"

# PowerShell
$env:FLASHDATA_API_KEY = "YOUR_API_KEY"

Search Google

import os
from flashdata import FlashData

with FlashData(api_key=os.getenv("FLASHDATA_API_KEY")) as client:
    response = client.google.search("coffee roasters", gl="us", hl="en", num=5)
    for page in response["results"]:
        for hit in page.get("organic", []):
            print(hit["title"], hit["link"])

Responses are ordinary dictionaries. response["results"] contains source payloads: Google web hits are in each payload's organic list. Other sources have their own fields. Credits charged for the request are in response["usage"]["credits"].

Read a YouTube caption track

import os
from flashdata import FlashData

with FlashData(api_key=os.getenv("FLASHDATA_API_KEY")) as client:
    response = client.youtube.captions("dQw4w9WgXcQ", language="en", auto=False)
    for track in response["results"]:
        print(track["languageCode"], track["isAutoGenerated"])
        for segment in track.get("segments", [])[:3]:
            print(segment["start"], segment["text"])

Captions default to an uploaded track (auto=False); transcripts default to an automatic track (auto=True). Choose a language and track that exist on the video. Missing tracks return an API error; the service does not generate or translate them.

Use asyncio

import os
import asyncio
from flashdata import AsyncFlashData

async def main():
    async with AsyncFlashData(api_key=os.getenv("FLASHDATA_API_KEY")) as client:
        response = await client.youtube.search("python tutorial", max_results=5)
        for payload in response["results"]:
            for video in payload.get("results", []):
                print(video["title"], video["videoId"])

asyncio.run(main())

AsyncFlashData provides the same resources. Use await for requests and async for for pagination and bulk row iterators. Reuse a client within an application or event loop.

Run work in the background

import os
from flashdata import FlashData

with FlashData(api_key=os.getenv("FLASHDATA_API_KEY")) as client:
    submitted = client.youtube.metadata("dQw4w9WgXcQ", mode="async", include_formats=False)
    job_id = submitted["id"]
    print("Save this job ID:", job_id, flush=True)
    client.jobs.wait(job_id, timeout=300, poll_interval=2)
    result = client.jobs.result(job_id)
    print(result)

Save the ID before waiting so the same job can be resumed after a process restart. wait() returns the completed job record; result() retrieves its JSON output. Use batch submission for mixed queries and bulk jobs for UTF-8 files containing one YouTube video ID per line.

Download YouTube videos in bulk

Create video-ids.txt with one video ID per line:

dQw4w9WgXcQ
jNQXAC9IVRw

Create private-storage.json with your destination configuration. For example, for S3:

{
  "storage_type": "s3",
  "storage_url": "your-media-bucket",
  "region": "us-east-1",
  "object_prefix": "youtube",
  "access_key": "YOUR_ACCESS_KEY",
  "secret_key": "YOUR_SECRET_KEY"
}

Then create the bulk download task, upload the ID file, and start processing:

import json
import os
from pathlib import Path
from flashdata import FlashData

storage = json.loads(Path("private-storage.json").read_text(encoding="utf-8"))

with FlashData(api_key=os.getenv("FLASHDATA_API_KEY")) as client:
    receipt = client.bulk.create({
        "source": "youtube_download",
        "options": {
            "download_type": "audio_video",
            "video_quality": "<=720",
            "output_format": "mp4",
            "storage": storage,
        },
    })
    bulk_id = receipt["job"]["id"]
    print("Save this bulk ID:", bulk_id, flush=True)
    client.bulk.upload_file(receipt, Path("video-ids.txt"))
    client.bulk.start(bulk_id)

    completed = client.bulk.wait(bulk_id, timeout=3600, poll_interval=5)
    print("Counts:", completed["counts"])
    for row in client.bulk.rows(bulk_id):
        print(row["video_id"], row["status"])

Media files are delivered to your configured bucket. bulk.rows() reads the task's JSON result records, including each video's status and source output in results. Check the final counts and row statuses: a completed task can contain failed items. Downloads are billed by actual delivered bytes; your storage provider may charge separately.

After a process restart or polling timeout, use the saved ID with client.bulk.get(), client.bulk.wait(), and client.bulk.rows() to continue reading the existing task. See bulk controls and recovery, download options, and all supported storage providers.

The runnable example saves the job record beside its NDJSON export and resumes an already started task without creating another submission:

python examples/bulk_download.py video-ids.txt --storage-file private-storage.json --output artifacts/downloads.ndjson
python examples/bulk_download.py --bulk-id bulk_YOUR_SAVED_ID --output artifacts/downloads.ndjson

Handle failures

import os
from flashdata import APIError, FlashData, SubmissionUnknownError

with FlashData(api_key=os.getenv("FLASHDATA_API_KEY"), timeout=120) as client:
    try:
        response = client.google.search("coffee", timeout=60)
    except APIError as error:
        print(error.status_code, error.code, error.request_id, error.retry_after)
    except SubmissionUnknownError:
        # The server may have accepted this paid request. Reconcile before resubmitting.
        raise

The SDK sends each request once. It does not automatically retry paid queries or mutations. See errors and recovery for rate limits, polling deadlines, failed jobs, and storage errors. Client configuration covers timeouts, proxies and TLS.

Supported APIs

Resource	Methods
Google	search, images, news, shopping, videos, places, bounded pages
YouTube	search, search_extended, metadata, transcript, captions, license_check, channel, suggestions, download
Queries	realtime, submit, batch with typed wire dictionaries
Jobs	get, wait, result
Bulk	create, upload_file, start, get, list, iter_jobs, update, pause, resume, cancel, wait, manifest, part, iter_rows, rows
Account	access, balance, pricing, limits, usage
Public	sources, plans, health, status (no key needed)

All 15 sources support background submission and batch. The 14 data sources also support realtime calls. Media downloads run in the background and require customer cloud storage. Bulk supports metadata, transcripts, captions, license labels, and downloads.

Guides and runnable examples

• Getting started — install, authenticate, understand responses.
• Source guides — parameters and useful recipes for every source.
• Examples — CLI inputs, readable output, JSON/CSV/NDJSON export.
• Pagination, jobs and batch, bulk datasets.
• Cloud storage, account and usage, API reference.

python examples/google_search.py "coffee roasters" --output artifacts/search.json
python examples/search_to_csv.py "coffee roasters" --max-pages 2 --output artifacts/search.csv
python examples/youtube_captions.py dQw4w9WgXcQ --language en
python examples/async_query.py "python tutorial" --output artifacts/result.json

These commands call the live API and consume credits. Run python examples/offline.py for a demonstration with a mock transport and no account.

The SDK comparison maps Python and TypeScript coverage, language differences, workflow guarantees, and verification limits side by side.

Install from source

From a checkout of this repository:

python -m pip install .

Development

python -m pip install -e ".[dev,docs]"
python -m pytest
python -m ruff check src tests examples scripts docs/conf.py
python -m ruff format --check src tests examples scripts docs/conf.py
python -m mypy src/flashdata
python -m sphinx -n -W --keep-going -b html docs docs/_build/html
python -m build
python scripts/verify_package.py

Tests use mock transports and a local HTTP server by default. Explicit opt-in tests exercise live APIs; see contributing. The project is MIT licensed. Report sensitive issues through the security policy.