Upwork Web Scraping: API Access, Python and Proxy Setup

python -m pip install requests

"""Export one authorized Upwork GraphQL search page, or process an offline fixture."""
import argparse
import csv
import json
import os
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path

import requests
from rola_connection import make_session

QUERY = '''query SearchJobs($filter: MarketplaceJobPostingsSearchFilter) {
  marketplaceJobPostingsSearch(
    marketPlaceJobFilter: $filter
    searchType: USER_JOBS_SEARCH
    sortAttributes: [{field: RECENCY}]
  ) {
    totalCount
    pageInfo { endCursor hasNextPage }
    edges { node {
      id title description ciphertext publishedDateTime
      amount { rawValue displayValue currency }
      hourlyBudgetMin { rawValue displayValue currency }
      hourlyBudgetMax { rawValue displayValue currency }
      skills { name }
    } }
  }
}'''
FIELDS = ['job_id', 'title', 'description', 'ciphertext', 'published_at',
          'amount_json', 'hourly_min_json', 'hourly_max_json', 'skills_json']


def export_page(payload, output):
    if payload.get('errors'):
        raise ValueError('GraphQL returned errors. Review scopes, token and query before retrying.')
    page = payload['data']['marketplaceJobPostingsSearch']
    rows, seen, counts = [], set(), Counter()
    for edge in page['edges']:
        node = edge['node']
        job_id = str(node['id'])
        if job_id in seen:
            continue
        seen.add(job_id)
        skills = [skill['name'] for skill in (node.get('skills') or [])]
        counts.update({skill.strip().casefold() for skill in skills if skill.strip()})
        rows.append(dict(job_id=job_id, title=node.get('title') or '',
                         description=node.get('description') or '',
                         ciphertext=node.get('ciphertext') or '',
                         published_at=node.get('publishedDateTime') or '',
                         amount_json=json.dumps(node.get('amount'), ensure_ascii=False),
                         hourly_min_json=json.dumps(node.get('hourlyBudgetMin'), ensure_ascii=False),
                         hourly_max_json=json.dumps(node.get('hourlyBudgetMax'), ensure_ascii=False),
                         skills_json=json.dumps(skills, ensure_ascii=False)))
    with output.open('w', encoding='utf-8-sig', newline='') as file:
        writer = csv.DictWriter(file, fieldnames=FIELDS)
        writer.writeheader()
        writer.writerows(rows)
    for skill, count in sorted(counts.items(), key=lambda item: (-item[1], item[0])):
        print(f'{skill}: {count}')
    return len(rows), page.get('pageInfo') or {}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--search', default='python')
    parser.add_argument('--after', default='0')
    parser.add_argument('--first', type=int, default=5)
    parser.add_argument('--output', type=Path, default=Path('api_jobs.csv'))
    parser.add_argument('--fixture', type=Path)
    parser.add_argument('--proxy', action='store_true')
    args = parser.parse_args()
    raw = args.output.with_suffix('.response.json')
    metadata = args.output.with_suffix('.meta.json')
    if args.first < 1 or args.first > 5:
        parser.error('This small-page example accepts --first from 1 to 5.')
    if args.fixture and args.fixture.resolve() in {p.resolve() for p in (args.output, raw, metadata)}:
        parser.error('Choose output paths different from the fixture.')
    if any(path.exists() for path in (args.output, raw, metadata)):
        parser.error('Choose a new output name to preserve earlier batches.')
    variables = {'filter': {'searchExpression_eq': args.search,
                            'pagination_eq': {'after': args.after, 'first': args.first}}}
    if args.fixture:
        payload = json.loads(args.fixture.read_text(encoding='utf-8'))
    else:
        token = os.environ.get('UPWORK_ACCESS_TOKEN')
        if not token:
            parser.error('Set UPWORK_ACCESS_TOKEN to your authorized OAuth access token.')
        with make_session(args.proxy) as session:
            response = session.post('https://api.upwork.com/graphql',
                                    headers={'Authorization': f'Bearer {token}'},
                                    json={'query': QUERY, 'variables': variables}, timeout=(10, 30))
            response.raise_for_status()
            payload = response.json()
    count, page_info = export_page(payload, args.output)
    raw.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding='utf-8')
    metadata.write_text(json.dumps({'mode': 'fixture' if args.fixture else 'api',
                                   'processed_at_utc': datetime.now(timezone.utc).isoformat(),
                                   'requested_variables': None if args.fixture else variables,
                                   'page_info': page_info}, indent=2), encoding='utf-8')
    print(f'Exported unique jobs: {count} -> {args.output.name}')
    print('Page info: ' + json.dumps(page_info))


if __name__ == '__main__':
    try:
        main()
    except (requests.RequestException, ValueError, KeyError, TypeError):
        raise SystemExit('Export failed. Check network access, API scopes and response structure; no automatic retry.')

$upworkToken = Read-Host 'Upwork OAuth access token' -AsSecureString
$env:UPWORK_ACCESS_TOKEN = [System.Net.NetworkCredential]::new('', $upworkToken).Password
python fetch_jobs.py --search "python" --first 5 --output api_jobs.csv
Remove-Item Env:UPWORK_ACCESS_TOKEN

python fetch_jobs.py --search "python" --after "CURSOR_FROM_PREVIOUS_RESPONSE" --first 5 --output api_jobs_page2.csv

{
  "data": {
    "marketplaceJobPostingsSearch": {
      "totalCount": 2,
      "pageInfo": {
        "endCursor": "demo-end",
        "hasNextPage": false
      },
      "edges": [
        {
          "node": {
            "id": "demo-1",
            "title": "Fictional hourly project",
            "description": "Offline demonstration only",
            "ciphertext": "demo-not-a-live-job",
            "publishedDateTime": null,
            "amount": null,
            "hourlyBudgetMin": {
              "displayValue": "15",
              "currency": "USD",
              "rawValue": "15"
            },
            "hourlyBudgetMax": {
              "displayValue": "45",
              "currency": "USD",
              "rawValue": "45"
            },
            "skills": [
              {
                "name": "Python"
              },
              {
                "name": "Data Scraping"
              },
              {
                "name": "Python"
              }
            ]
          }
        },
        {
          "node": {
            "id": "demo-2",
            "title": "Fictional fixed-price project",
            "description": "Offline demonstration only",
            "ciphertext": "demo-not-a-live-job-2",
            "publishedDateTime": null,
            "amount": {
              "displayValue": "500",
              "currency": "USD",
              "rawValue": "500"
            },
            "hourlyBudgetMin": null,
            "hourlyBudgetMax": null,
            "skills": [
              {
                "name": "Python"
              },
              {
                "name": "API Integration"
              }
            ]
          }
        }
      ]
    }
  }
}

python fetch_jobs.py --fixture demo_api_response.json --output demo_api_jobs.csv

python: 2
api integration: 1
data scraping: 1
Exported unique jobs: 2 -> demo_api_jobs.csv
Page info: {"endCursor": "demo-end", "hasNextPage": false}

"""Optional Rola HTTP proxy configuration and an independent exit-IP check."""
import ipaddress
import os
from urllib.parse import quote

import requests


def make_session(use_proxy=False):
    session = requests.Session()
    session.trust_env = False
    if use_proxy:
        values = {key: os.environ.get('ROLA_' + key, '')
                  for key in ('HOST', 'PORT', 'USERNAME', 'PASSWORD')}
        if not all(values.values()):
            raise ValueError('Set ROLA_HOST, ROLA_PORT, ROLA_USERNAME and ROLA_PASSWORD.')
        host, port = values['HOST'], values['PORT']
        if not host or any(c not in 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.-' for c in host):
            raise ValueError('ROLA_HOST must be the assigned gateway hostname or IPv4 address.')
        if not port.isdigit() or not 1 <= int(port) <= 65535:
            raise ValueError('ROLA_PORT must be a valid assigned port.')
        user = quote(values['USERNAME'], safe='')
        password = quote(values['PASSWORD'], safe='')
        proxy = f'http://{user}:{password}@{host}:{port}'
        session.proxies.update(http=proxy, https=proxy)
    return session


if __name__ == '__main__':
    try:
        with make_session(True) as session:
            response = session.get('https://api.ipify.org?format=json', timeout=(10, 30))
            response.raise_for_status()
            address = ipaddress.ip_address(response.json()['ip'])
            print(f'Proxy exit IP: {address}')
    except (requests.RequestException, ValueError, KeyError):
        raise SystemExit('Connection check failed. Review proxy settings and network access.')

$env:ROLA_HOST = 'YOUR_ASSIGNED_GATEWAY'
$env:ROLA_PORT = 'YOUR_ASSIGNED_HTTP_PORT'
$env:ROLA_USERNAME = 'YOUR_COMPLETE_PROXY_USERNAME'
$rolaPassword = Read-Host 'Rola proxy password' -AsSecureString
$env:ROLA_PASSWORD = [System.Net.NetworkCredential]::new('', $rolaPassword).Password
python rola_connection.py

python fetch_jobs.py --search "python" --first 5 --proxy --output api_jobs_proxy.csv

Remove-Item Env:ROLA_PASSWORD

"""Normalize a local CSV. --demo accepts example.com fixture URLs only."""
import argparse
import csv
import re
from decimal import Decimal, InvalidOperation
from pathlib import Path
from urllib.parse import urlsplit

INPUT_FIELDS = {'title', 'budget', 'budget_unit', 'skills', 'date_posted', 'url'}
FIELDS = ['title', 'budget_raw', 'budget_min_usd', 'budget_max_usd',
          'budget_unit', 'budget_status', 'skills', 'date_posted', 'url', 'source_url']


def job_url(value, demo=False):
    try:
        parts = urlsplit(value.strip())
        host = 'example.com' if demo else 'www.upwork.com'
        match = re.fullmatch(r'/jobs/[A-Za-z0-9-]+_(~\d+)/?', parts.path)
        if parts.scheme != 'https' or parts.netloc != host or not match:
            return None
        return match[1], f'https://{host}{parts.path.rstrip("/")}/'
    except ValueError:
        return None


def budget_range(value, unit):
    match = re.fullmatch(r'USD\s+(\d+(?:\.\d+)?)(?:\s*[-–]\s*(\d+(?:\.\d+)?))?', value.strip())
    if unit not in {'HOUR', 'FIXED'} or not match:
        return '', '', 'unparsed'
    try:
        low = Decimal(match[1])
        high = Decimal(match[2] or match[1])
        if not low.is_finite() or not high.is_finite() or low < 0 or high < low:
            return '', '', 'unparsed'
        return str(low), str(high), 'parsed'
    except InvalidOperation:
        return '', '', 'unparsed'


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('input', type=Path)
    parser.add_argument('--output', type=Path, default=Path('upwork_jobs_analysis.csv'))
    parser.add_argument('--demo', action='store_true')
    args = parser.parse_args()
    if args.input.resolve() == args.output.resolve():
        parser.error('Use a different output path to preserve the source CSV.')
    with args.input.open(encoding='utf-8-sig', newline='') as file:
        reader = csv.DictReader(file)
        if not INPUT_FIELDS.issubset(reader.fieldnames or []):
            parser.error('Input must include: ' + ', '.join(sorted(INPUT_FIELDS)))
        source = list(reader)
    rows, seen = [], set()
    invalid = duplicate = unparsed = 0
    for item in source:
        item = {key: (item.get(key) or '').strip() for key in INPUT_FIELDS}
        link = job_url(item['url'], args.demo)
        if not link:
            invalid += 1
            continue
        job_id, url = link
        if job_id in seen:
            duplicate += 1
            continue
        seen.add(job_id)
        unit = item['budget_unit'].upper()
        low, high, status = budget_range(item['budget'], unit)
        unparsed += status == 'unparsed'
        rows.append(dict(title=item['title'], budget_raw=item['budget'],
                         budget_min_usd=low, budget_max_usd=high,
                         budget_unit=unit, budget_status=status,
                         skills=item['skills'], date_posted=item['date_posted'],
                         url=url, source_url=item['url']))
    with args.output.open('w', encoding='utf-8-sig', newline='') as file:
        writer = csv.DictWriter(file, fieldnames=FIELDS)
        writer.writeheader()
        writer.writerows(rows)
    print(f'Source rows: {len(source)}')
    print(f'Saved rows: {len(rows)} -> {args.output.name}')
    print(f'Skipped: invalid URL format={invalid}, duplicate job ID={duplicate}')
    print(f'Budgets requiring review: {unparsed}')


if __name__ == '__main__':
    main()

"""Validate normalized records and count each skill once per job."""
import argparse
import csv
from collections import Counter
from decimal import Decimal, InvalidOperation
from pathlib import Path
from analyze_jobs import FIELDS, budget_range, job_url


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('input', type=Path)
    parser.add_argument('--skills-out', type=Path, default=Path('skill_frequency.csv'))
    parser.add_argument('--demo', action='store_true')
    args = parser.parse_args()
    if args.input.resolve() == args.skills_out.resolve():
        parser.error('Use a different output path.')
    with args.input.open(encoding='utf-8-sig', newline='') as file:
        reader = csv.DictReader(file)
        if not set(FIELDS).issubset(reader.fieldnames or []):
            parser.error('Missing normalized columns.')
        rows = list(reader)
    seen, counts = set(), Counter()
    pending = 0
    for line, row in enumerate(rows, 2):
        link = job_url(row['url'] or '', args.demo)
        original = job_url(row['source_url'] or '', args.demo)
        if not link or not original or link[0] != original[0] or link[0] in seen:
            parser.error(f'Invalid or duplicate job reference at row {line}.')
        seen.add(link[0])
        expected = budget_range(row['budget_raw'] or '', row['budget_unit'] or '')
        actual = (row['budget_min_usd'], row['budget_max_usd'], row['budget_status'])
        if actual != expected:
            parser.error(f'Budget fields do not match the original value at row {line}.')
        if row['budget_status'] == 'parsed':
            try:
                low, high = Decimal(row['budget_min_usd']), Decimal(row['budget_max_usd'])
                valid = low.is_finite() and high.is_finite() and 0 <= low <= high
            except (InvalidOperation, ValueError):
                valid = False
            if not valid:
                parser.error(f'Invalid numeric bounds at row {line}.')
        else:
            pending += 1
        counts.update({skill.strip().casefold() for skill in (row['skills'] or '').split(',') if skill.strip()})
    with args.skills_out.open('w', encoding='utf-8-sig', newline='') as file:
        writer = csv.writer(file)
        writer.writerow(['skill', 'job_count'])
        writer.writerows(sorted(counts.items(), key=lambda pair: (-pair[1], pair[0])))
    print('Record checks: PASS' if rows else 'Record checks: no records to validate')
    print(f'Unique jobs: {len(rows)} | Distinct skills: {len(counts)}')
    print(f'Budgets requiring review: {pending}')
    for skill, count in sorted(counts.items(), key=lambda pair: (-pair[1], pair[0])):
        print(f'{skill}: {count}')
    print(f'Saved skill counts -> {args.skills_out.name}')


if __name__ == '__main__':
    main()

title,budget,budget_unit,skills,date_posted,url
Demo hourly extraction project,USD 15-45,HOUR,"Python, Data Scraping, Python",,https://example.com/jobs/Demo-extraction_~1001/
Demo API integration project,USD 25-50,HOUR,"Python, API Integration",,https://example.com/jobs/Demo-api_~1002/
Demo fixed-price cleanup project,USD 500,FIXED,"Data Scraping, Data Cleaning",,https://example.com/jobs/Demo-cleanup_~1003/
Demo incomplete reference,USD 20-30,HOUR,Python,,https://example.com/jobs/
Demo duplicate with tracking,USD 15-45,HOUR,"Python, Data Scraping",,https://example.com/jobs/Demo-extraction_~1001/?source=demo

python analyze_jobs.py demo_jobs.csv --demo --output upwork_jobs_analysis.csv
python inspect_results.py upwork_jobs_analysis.csv --demo --skills-out skill_frequency.csv

Source rows: 5
Saved rows: 3 -> upwork_jobs_analysis.csv
Skipped: invalid URL format=1, duplicate job ID=1
Budgets requiring review: 0

Record checks: PASS
Unique jobs: 3 | Distinct skills: 4
Budgets requiring review: 0
data scraping: 2
python: 2
api integration: 1
data cleaning: 1
Saved skill counts -> skill_frequency.csv