Paginate search results

import json
import os
import sys
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
from uuid import UUID

BODY = {
    'field_ids': ['identifier', 'short_description', 'categories', 'location_identifiers'],
    'query': [
        {'type': 'predicate', 'field_id': 'categories', 'operator_id': 'includes',
         'values': ['5c4e6926-5ff7-b188-0892-c8eb036c5ace']},
        {'type': 'predicate', 'field_id': 'location_identifiers', 'operator_id': 'includes',
         'values': ['528f5e3c-90d1-1111-5d1c-2e4ff979d58e']},
    ],
    'limit': 3,
}


def fetch(body, key):
    request = Request(
        'https://api.crunchbase.com/v4/data/searches/organizations',
        data=json.dumps(body).encode(),
        headers={'X-cb-user-key': key, 'Content-Type': 'application/json',
                 'Accept': 'application/json'})
    with urlopen(request, timeout=30) as response:
        data = json.load(response)
    entities = data.get('entities') if isinstance(data, dict) else None
    if not isinstance(entities, list):
        raise ValueError('Expected an entities array.')
    for entity in entities:
        uid = entity.get('uuid') if isinstance(entity, dict) else None
        if not isinstance(uid, str):
            raise ValueError('Result missing its UUID.')
        UUID(uid)
    return entities


def main():
    key = os.environ.get('CRUNCHBASE_API_KEY', '').strip()
    if not key:
        raise ValueError('Set CRUNCHBASE_API_KEY first.')
    cursor = None
    for page_number in (1, 2):
        body = dict(BODY)
        if cursor is not None:
            body['after_id'] = cursor
        entities = fetch(body, key)
        print(json.dumps({'page': page_number, 'entities': entities}))
        if not entities:
            print('End reached: this request returned an empty page.')
            return
        next_cursor = entities[-1]['uuid']
        if next_cursor == cursor:
            raise ValueError('Repeated cursor; stop and inspect the response.')
        cursor = next_cursor
    print('Two-page sample complete. More results may remain; this is not a full export.')


if __name__ == '__main__':
    try:
        main()
    except HTTPError as error:
        print(f'Stopped: HTTP {error.code}. Check the error before retrying.', file=sys.stderr)
        sys.exit(1)
    except (URLError, TimeoutError):
        print('Stopped: network error or timeout.', file=sys.stderr)
        sys.exit(1)
    except ValueError as error:
        print(f'Stopped: {error}', file=sys.stderr)
        sys.exit(1)

import argparse
import json
import os
import random
import sys
import time
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
from uuid import UUID

URL = 'https://api.crunchbase.com/v4/data/searches/organizations'
QUERY = {
    'field_ids': ['identifier', 'short_description', 'categories', 'location_identifiers'],
    'query': [
        {'type': 'predicate', 'field_id': 'categories', 'operator_id': 'includes',
         'values': ['5c4e6926-5ff7-b188-0892-c8eb036c5ace']},
        {'type': 'predicate', 'field_id': 'location_identifiers', 'operator_id': 'includes',
         'values': ['528f5e3c-90d1-1111-5d1c-2e4ff979d58e']},
    ],
    'limit': 3,
}


def retry_delay(value):
    if value is None:
        return 0
    try:
        return max(0, float(value))
    except ValueError:
        try:
            date = parsedate_to_datetime(value)
            return max(0, (date - datetime.now(timezone.utc)).total_seconds())
        except (ValueError, TypeError, OverflowError):
            return 0


def fetch_page(body, key):
    # Retry this same page; never advance the cursor after a failed request.
    for attempt in range(4):
        time.sleep(0.4)  # Pace every attempt, including retries, in this process.
        req = Request(URL, data=json.dumps(body).encode(), headers={
            'X-cb-user-key': key, 'Content-Type': 'application/json',
            'Accept': 'application/json',
        })
        wait = 0
        try:
            with urlopen(req, timeout=30) as response:
                data = json.load(response)
            if not isinstance(data, dict) or not isinstance(data.get('entities'), list):
                raise ValueError('Expected a JSON object containing an entities array.')
            return data['entities']
        except HTTPError as error:
            status = error.code
            wait = retry_delay(error.headers.get('Retry-After'))
            error.close()
            if status != 429 and not 500 <= status <= 599:
                raise RuntimeError(f'HTTP {status}; correct the request or access before retrying.') from None
            reason = f'HTTP {status}'
        except (URLError, TimeoutError):
            reason = 'network error or timeout'
        if attempt == 3:
            raise RuntimeError(f'{reason}; exhausted four attempts for this page.')
        # Stop instead of retrying early when the server requests a long pause.
        if wait > 60:
            raise RuntimeError(f'{reason}; server requests a pause of {wait:.0f}s. Wait before restarting.')
        time.sleep(max(wait, 2 ** attempt + random.uniform(0, 0.5)))


def traverse(fetch, max_pages, emit):
    seen_ids, seen_cursors = set(), set()
    cursor = None
    duplicates = 0
    for page_number in range(1, max_pages + 1):
        body = dict(QUERY)
        if cursor is not None:
            body['after_id'] = cursor
        entities = fetch(body)
        if not isinstance(entities, list):
            raise ValueError('Expected an entities array, not a missing or null value.')
        if not entities:
            return {'status': 'end_reached', 'pages_requested': page_number,
                    'unique_entities': len(seen_ids), 'duplicates_skipped': duplicates}
        # Validate the whole page before emitting any of its records.
        ids = []
        for entity in entities:
            uid = entity.get('uuid') if isinstance(entity, dict) else None
            if not isinstance(uid, str):
                raise ValueError('Result missing its UUID; cursor cannot be trusted.')
            UUID(uid)
            ids.append(uid)
        next_cursor = ids[-1]  # Last raw result, even if already seen.
        if next_cursor in seen_cursors:
            raise RuntimeError('Repeated cursor; stopped to prevent an infinite loop.')
        for uid, entity in zip(ids, entities):
            if uid in seen_ids:
                duplicates += 1
                continue
            emit(entity)
            seen_ids.add(uid)
        seen_cursors.add(next_cursor)
        cursor = next_cursor
    raise RuntimeError(f'Reached max-pages={max_pages} before an empty page; output is partial.')


def main():
    parser = argparse.ArgumentParser(description='Page through SaaS organizations in San Francisco.')
    parser.add_argument('--max-pages', type=int, default=10,
                        help='Safety cap including the final empty-page request (default: 10).')
    args = parser.parse_args()
    if args.max_pages < 1:
        parser.error('--max-pages must be positive')
    key = os.environ.get('CRUNCHBASE_API_KEY', '').strip()
    if not key:
        raise ValueError('Set CRUNCHBASE_API_KEY to your provisioned key first.')
    summary = traverse(lambda body: fetch_page(body, key), args.max_pages,
                       lambda entity: print(json.dumps(entity), flush=True))
    print(json.dumps(summary), file=sys.stderr)


if __name__ == '__main__':
    try:
        main()
    except (ValueError, RuntimeError) as error:
        print(f'INCOMPLETE: {error} Any records already written are partial.', file=sys.stderr)
        sys.exit(2)