How to Use KnoxProxy with Apache Airflow

from datetime import datetime, timedelta
import requests
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.hooks.base import BaseHook

PROXY_CONN_ID = "knoxproxy"

def fetch_with_proxy(url: str, country: str = "", **kwargs):
    """Fetch a URL through KnoxProxy."""
    conn = BaseHook.get_connection(PROXY_CONN_ID)
    username = conn.login
    if country:
        username = f"{username}-country-{country}"
    proxy_url = f"http://{username}:{conn.password}@{conn.host}:{conn.port}"

    proxies = {"http": proxy_url, "https": proxy_url}
    response = requests.get(url, proxies=proxies, timeout=30)
    response.raise_for_status()

    # Push result to XCom for downstream tasks
    kwargs["ti"].xcom_push(key="response", value=response.text[:5000])
    return response.status_code

with DAG(
    dag_id="proxied_scraper",
    start_date=datetime(2026, 1, 1),
    schedule_interval=timedelta(hours=6),
    catchup=False,
    tags=["scraping", "proxy"],
) as dag:

    scrape_us = PythonOperator(
        task_id="scrape_us_endpoint",
        python_callable=fetch_with_proxy,
        op_kwargs={"url": "https://httpbin.org/ip", "country": "us"},
    )

    scrape_de = PythonOperator(
        task_id="scrape_de_endpoint",
        python_callable=fetch_with_proxy,
        op_kwargs={"url": "https://httpbin.org/ip", "country": "de"},
    )

    scrape_us >> scrape_de