Issue 1:
Apache Airflow version

2.3.2 (latest released)

What happened

Using the provided minimal example, task_instance (i.e., ti) is accessible when using PythonOperator but fails when using PythonVirtualenvOperator.

What you think should happen instead

I would expect PythonOperator and PythonVirtualenvOperator to behave in a similar way. Instead, none of the xcom examples work with PythonVirtualenvOperator.

How to reproduce

Switch between PythonOperator and PythonVirtualenvOperator. Using PythonVirtualenvOperator dag will fail.

from airflow import DAG
from airflow.operators.python import PythonVirtualenvOperator, PythonOperator
from datetime import datetime, timedelta

def _pusher(ti):
ti.xcom_push(key='my-key', value='my-value')

def _puller(ti):
value = ti.xcom_pull(task_ids='pusher', key='my-key')
assert value == 'my-value'

with DAG('minimal-example', schedule_interval='@once', start_date=datetime(2021, 1, 1), catchup=False) as dag:
#pusher = PythonVirtualenvOperator(
pusher = PythonOperator(
task_id='pusher',
python_callable=_pusher,
#requirements=[],
)

#puller = PythonVirtualenvOperator(
puller = PythonOperator(
task_id='puller',
python_callable=_puller,
#requirements=[],
)

pusher >> puller

Operating System

Ubuntu 20.04

Versions of Apache Airflow Providers

Apache airflow 2.3.2 in docker container

Deployment

Docker-Compose

Deployment details

https://github.com/apache/airflow/blob/2.3.2/docs/apache-airflow/start/docker-compose.yaml

Anything else

No response

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct

Issue 2:
Apache Airflow version: 1.10.13

Environment: Docker (Ubuntu 18.4 - Python 3.7)

Cloud provider or hardware configuration: Azure

OS (e.g. from /etc/os-release): Docker (Ubuntu 18.4 - Python 3.7)

Kernel (e.g. uname -a): Docker (Ubuntu 18.4 - Python 3.7)

Install tools: N/A

Others: N/A

What happened:

When we enable provide_context=True for PythonVirtualenvOperator and try to use the xcom_push to pass a variable I get this error:

File "/tmp/venv0upgqome/script.py", line 13, in push\n kwargs[\'ti\'].xcom_push(key=\'value from pusher 1\', value=value_1)\nKeyError: \'ti\'\n'

How to reproduce it:

import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonVirtualenvOperator

args = {
'owner': 'Airflow',
'start_date': airflow.utils.dates.days_ago(2),

}

dag = DAG('BAtatas', schedule_interval="@once", default_args=args)

def push(**kwargs):
"""Pushes an XCom without a specific target"""
value_1 = [1, 2, 3]
print("printing the kwargs!!!")
print(kwargs)
kwargs['ti'].xcom_push(key='value from pusher 1', value=value_1)

def push_by_returning(**kwargs):
value_2 = {'a': 'b'}
"""Pushes an XCom without a specific target, just by returning it"""
return value_2

def puller(**kwargs):
value_2 = {'a': 'b'}
value_1 = [1, 2, 3]
"""Pull all previously pushed XComs and check if the pushed values match the pulled values."""
ti = kwargs['ti']

# get value_1
pulled_value_1 = ti.xcom_pull(key=None, task_ids='push')
assert pulled_value_1 == value_1

# get value_2
pulled_value_2 = ti.xcom_pull(task_ids='push_by_returning')
assert pulled_value_2 == value_2

# get both value_1 and value_2
pulled_value_1, pulled_value_2 = ti.xcom_pull(
key=None, task_ids=['push', 'push_by_returning'])
assert (pulled_value_1, pulled_value_2) == (value_1, value_2)

push1 = PythonVirtualenvOperator(
task_id='push',
dag=dag,
python_callable=push,
requirements=[],
python_version='3.7',
use_dill=False,
provide_context=True,
system_site_packages=True,

)

push2 = PythonVirtualenvOperator(
task_id='push_by_returning',
dag=dag,
python_callable=push_by_returning,
requirements=[],
python_version='3.7',
use_dill=False,
provide_context=True,
system_site_packages=True,

)

pull = PythonVirtualenvOperator(
task_id='puller',
dag=dag,
python_callable=puller,
requirements=[],
python_version='3.7',
use_dill=False,
provide_context=True,
system_site_packages=True,

)

pull << [push1, push2]