[Python] Airflow Send Request is Stay Pending

I have Airflow dag.

@task
def check_and_prepare_data():
import pydevd_pycharm
pydevd_pycharm.settrace('localhost', port=5678, stdoutToServer=True, stderrToServer=True, suspend=False)
logger = logging.getLogger("airflow.task")

try:
logger.info("[check_and_prepare_data] Starting DAG task...")

base_url = get_variable("WORKFLOW_API_URL_TEMPLATE")
api_url = f"{base_url}/pending"
logger.info(f"[check_and_prepare_data] API URL composed: {api_url}")

logger.info("[check_and_prepare_data] Getting Keycloak token...")
token = get_keycloak_token()


def get_variable(variable_name: str) -> str:
logger.info(f"[get_variable] Fetching variable '{variable_name}' from Airflow...")
return Variable.get(variable_name)


Below code is aim to get token from keycloak.

def get_keycloak_token() -> str:
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

try:
logger.info("[get_keycloak_token] Retrieving Keycloak configuration from Airflow Variables...")
keycloak_base_url = Variable.get("keycloak_base_url")
realm = Variable.get("keycloak_realm")
token_endpoint = f"/realms/{realm}/protocol/openid-connect/token"
token_url = urljoin(keycloak_base_url, token_endpoint)

client_id = Variable.get("keycloak_client_id")
username = Variable.get("keycloak_username")
password = Variable.get("keycloak_password")

logger.info(f"[get_keycloak_token] Base URL: {keycloak_base_url}")
logger.info(f"[get_keycloak_token] Realm: {realm}")
logger.info(f"[get_keycloak_token] Token URL: {token_url}")
logger.info(f"[get_keycloak_token] Client ID: {client_id}")
logger.info(f"[get_keycloak_token] Username: {username}")
logger.info(f"[get_keycloak_token] Password: {'***' if password else 'None'}")

logger.info("[get_keycloak_token] Sending token request to Keycloak...")

token_data = {
'grant_type': 'password',
'username': username,
'password': password,
'client_id': client_id,
}

headers = {"Content-Type": "application/x-www-form-urlencoded"}

response = requests.post(
token_url,
data=token_data,
headers=headers,
timeout=30,
verify=False
)


When i send curl request, i can obtain token. When i send in venv, i can obtain token. But when i trigger this task, code is wait in in the request.post()... section and my task stay running status.

How can i fix this.?

Continue reading...