@task.virtualenv doesn't show logs #54706

import logging
logging.info("This is a simple log")

logging.info(f"Output directory: {output_dir}")
logging.info(f"Temporary directory: {temp_dir}")

# Example: @task
@task
def task_one(config_path: str, logical_date=None, params=None, ti=None):
    import pandas as pd  # Import inside task per best practices
    import json
    import csv
    from pathlib import Path
    import logging

    try:
        with open(config_path) as f:
            config = json.load(f)

        task_id = ti.task_id if ti else "unknown_task"

        temp_dir = Path(config.get("temp_dir"))
        raw_data = config.get("raw_data", 0)
        processed_data = raw_data * 2

        temp_dir.mkdir(parents=True, exist_ok=True)

        csv_file = temp_dir / "processed_data.csv"
        with open(csv_file, mode="w", newline="") as f:
            writer = csv.DictWriter(
                f, fieldnames=["logical_date", "task_id", "processed_data"]
            )
            writer.writeheader()
            writer.writerow(
                {
                    "logical_date": str(logical_date),
                    "task_id": task_id,
                    "processed_data": processed_data,
                }
            )

        logging.info(f"Processed data saved to {csv_file}")

    except Exception as e:
        logging.error(f"Error in @task: {e}")

    return


# Example: @task.virtualenv
@task.virtualenv(
    python_version="3.9",
    requirements="requirements_example.txt",
    system_site_packages=True,
    venv_cache_path="venvs/example",
    inherit_env=True,
)
def task_two(config_path: str, logical_date=None, params=None):
    import json
    import pendulum
    import os
    import csv
    from pathlib import Path
    import logging

    try:
        with open(config_path) as f:
            config = json.load(f)

        output_dir = Path(config.get("output_dir"))
        temp_dir = Path(config.get("temp_dir"))
        output_dir.mkdir(parents=True, exist_ok=True)

        logging.info(f"Output directory: {output_dir}")
        logging.info(f"Temporary directory: {temp_dir}")

        processed_csv = temp_dir / "processed_data.csv"
        if not processed_csv.exists():
            raise FileNotFoundError(f"Processed data file not found: {processed_csv}")

        with open(processed_csv, mode="r", newline="") as f:
            reader = csv.DictReader(f)
            processed_rows = [row for row in reader]

        logging.info(f"Read {len(processed_rows)} rows from processed data")

        output_file = output_dir / f"output_{logical_date}.csv"
        with open(output_file, mode="w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=processed_rows[0].keys())
            writer.writeheader()
            for row in processed_rows:
                writer.writerow(row)

        logging.info(f"Output data saved to {output_file}")

    except Exception as e:
        logging.error(f"Error in @task.virtualenv: {e}")

    return

from airflow.sdk import dag, Param


@dag(
    dag_id="example", 
    schedule = None,
    params={
        "Greeting": Param("Hello!", type="string")
    }
    # https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/params.html#params
)
def example():
    from utils.example import task_one, task_two

    config_path = "configs/config_example.json"
    task_one = task_one(config_path = config_path)
    task_two = task_two(config_path = config_path)

    task_one >> task_two

example()