Serialization issues with external_python decorator

[Snaptonis]

Hey everybody,

I started changing my dags to use @task.external_python().
I am still running into issues and I don't know what to do anymore.

A couple dags fail with:
PicklingError: Can't pickle <function get_connection at 0x7f105f947110>: it's not the same object as tasks.connections.get_connection.
Another fails with:
PicklingError: Can't pickle local object <function my_dag.<locals>.get_config at 0x7f105f94ba00>.

I have a mixture of normal tasks and external_python tasks.
I think the the problems start when I have expanded tasks where the result (so the xcom) is used in another task but I'm not certain.

Airflow Version: 3.3.0

If you need more infos I'm happy to provide.

Edit: More info
I will describe the flow for how the errors happen:
First error:

1. task()s that get some connections from airflow
2. task.external_python() that reads something from database
3. task.external_python() thats expanded with partial args that takes the connection and the data read from the previous task
4. task.external_python() that takes a different connection and the return value from the previous task
5. This task fails but it manages to do what it is supposed to do. The function mentioned in the error is not part of the task.

Second error:
Similar here.

1. taks()s that read connections and params for dag (get_config)
2. task.external_python() expanded with partial taking the values returned from the previous tasks
3. task.external_python() taking a connection and the values returned from the previous tasks
4. This task fails. Notably the function mentioned in the error is not part of this task.

[christianalberto]

Here is the straight-to-the-point explanation: when you use @task.external_python, Airflow tries to pickle your function to send it over to a separate Python environment. If the function tries to grab variables, helpers, or imports from outside its scope, pickling breaks.

To fix it, treat your task as a self-contained box. Everything it needs must live inside the function.

❌ Code that fails

from tasks.connections import get_connection  # ❌ Import outside task

def my_dag():

    def get_config():  # ❌ Local nested function outside task
        return {"env": "prod"}

    @task.external_python(python="/path/to/venv/bin/python")
    def my_task():
        cfg = get_config()  # 💥 Fails: cannot pickle local function
        conn = get_connection()  # 💥 Fails: module import mismatch across venvs

✅ Code fixed

def my_dag():

    @task.external_python(python="/path/to/venv/bin/python")
    def my_task():
        # ✅ Imports go INSIDE the task
        from tasks.connections import get_connection

        # ✅ Helper logic goes INSIDE the task
        cfg = {"env": "prod"}
        conn = get_connection()

3 golden rules to keep in mind:

1. Imports inside: Put any imports needed by the task directly inside the decorated function.
2. No nested functions: Don't reference helper functions defined inside my_dag().
3. Simple XComs: If you use .expand() or pass data between tasks, stick to plain data structures (strings, numbers, dicts, lists)—avoid passing complex custom objects or classes.

Moving the imports and helper logic inside the function will resolve both pickling errors right away.

[Snaptonis]

Thank you for your answer.
Unfortunately I have already made sure that all imports are local and no functions that dont exist in the scope are called.
The XComs are also fine.

I have updated my original post with some more info, if you would like to take a look again.

[yudin-s]

After your update, I would look less at the body of the failing external_python task and more at what is being passed into it.

For @task.external_python, Airflow still serializes the callable arguments before it starts the external interpreter. In the current operator code, the args/kwargs are written with the selected serializer as roughly:

{"args": self.op_args, "kwargs": resolved_kwargs}

The docs also call out two separate constraints:

• the function body is cut out into a temporary file, so imports must be inside the function and globals should not be relied on;
• op_args, op_kwargs, and return values still need to be serializable between the Airflow process and the external Python process.

That matches the clue you added: the function named in the PicklingError is not part of the task that fails. It can still be inside an upstream task's return value, a config object, a connection wrapper, a partially-applied argument, or something captured in an expanded/partial mapped argument. Airflow may be failing while serializing the arguments for the external task, not while running the function body.

I would narrow it this way:

1. Do not pass Airflow Connection objects, hooks, callables, modules, or rich config objects through XCom into an external task. Pass only primitives: strings, numbers, booleans, lists, and dicts of those.

2. For connections, pass a conn_id or a plain dict of exactly the fields you need:

@task
def get_config():
    return {
        "source_conn_id": "my_source",
        "target_conn_id": "my_target",
    }

Then resolve/import inside the external task:

@task.external_python(python="/path/to/venv/bin/python")
def load_data(config, rows):
    from airflow.hooks.base import BaseHook

    source = BaseHook.get_connection(config["source_conn_id"])
    # or import your own DB client and use primitive host/user/schema fields

If the external environment does not have the same Airflow version installed, avoid BaseHook inside it and pass primitive connection fields instead.

3. Check .partial(...) and .expand(...) inputs specifically. A non-serializable object in a partial kwarg can fail later in a downstream mapped external task, which makes the error look unrelated.

4. The Can't pickle local object my_dag.<locals>.get_config error usually means a function object itself is being passed or captured somewhere. Make sure you are passing get_config() / XComArg results, not the get_config function object, and that any returned config dict does not contain callables.

5. If you truly need to pass a more complex but still normal Python object, try serializer="cloudpickle" and install cloudpickle where Airflow expects it. But I would not use that to pass live Airflow objects; it is better to make the task boundary plain-data-only.

So the practical rule is: keep all imports inside the external function, but also make every upstream XCom, mapped argument, and partial kwarg crossing into external_python plain data. The error naming get_connection / get_config is probably pointing at one of those values being serialized, not necessarily code executed by the failing task.

If this points you to the bad mapped/partial value, please mark it as answered so other external_python serialization cases can find the argument-boundary diagnosis.

[Snaptonis]

Thank you for your answer.

All tasks return either a dict, list[dict] or bool, so I don't think thats it.
I have tried cloudpickle as well as dill. They also had problems although partially different errors. So I figured if all serializers struggle I am doing something wrong and returned to the default one.

I guess I could try not mixing tasks and taks.external_python?
It seems a bit odd to me using airflow imports in external_python, tho.

[latent-9]

@task.external_python runs your callable in a separate interpreter, so everything it touches has to survive being serialized and re-imported there. The two errors point at two separate causes.

can't pickle local object comes from defining helpers or closures inside the DAG or task function. Locals are pickled by value and nested functions cannot be, so move get_config and any helpers into a module that is importable inside the external environment and import them normally.

The connection error comes from touching Airflow objects inside the external task. get_connection and Connection belong to the Airflow runtime, which the external interpreter does not share, so the identity check fails. The fix is to not use Airflow connections inside an external_python task. Fetch the connection in a normal @task, pull out the plain values you need, and pass those primitives in. Keep external_python inputs and outputs to picklable data only.

If you need the venv isolation together with Airflow context, @task.virtualenv shares more of the runtime, or a cloudpickle based serializer is the usual alternative.

[Snaptonis]

Thank you for your answer.

Regarding can't pickle local object, get_config is a @task() that returns a dict, where the values are read from the dag params. The resulting XCom is then passed to an external_python task that is expanded with partial args. So far so good. But the next task, which is also external and uses the XCom created by the prior task, fails. So the tasks never use get_config() but only the result of it.

Same goes for the connection, the @task() only returns a dict that is used by following tasks.

[mohammadrezwankhan]

[Snaptonis]

Full trace:
Task failed with exception source=task loc=task_runner.py:1673
PicklingError: Can't pickle <function get_connection at 0x7f105f947110>: it's not the same object as tasks.connections.get_connection

File "/home/airflow/.local/lib/python3.14/site-packages/airflow/sdk/execution_time/task_runner.py", line 1576 in run

File "/home/airflow/.local/lib/python3.14/site-packages/airflow/sdk/execution_time/task_runner.py", line 189 in wrapper

File "/home/airflow/.local/lib/python3.14/site-packages/airflow/sdk/execution_time/task_runner.py", line 2125 in _execute_task

File "/home/airflow/.local/lib/python3.14/site-packages/airflow/sdk/bases/operator.py", line 445 in wrapper

File "/home/airflow/.local/lib/python3.14/site-packages/airflow/sdk/bases/decorator.py", line 398 in execute

File "/home/airflow/.local/lib/python3.14/site-packages/airflow/sdk/bases/operator.py", line 445 in wrapper

File "/home/airflow/.local/lib/python3.14/site-packages/airflow/providers/standard/operators/python.py", line 551 in execute

File "/home/airflow/.local/lib/python3.14/site-packages/airflow/sdk/bases/operator.py", line 445 in wrapper

File "/home/airflow/.local/lib/python3.14/site-packages/airflow/providers/standard/operators/python.py", line 231 in execute

File "/home/airflow/.local/lib/python3.14/site-packages/airflow/providers/standard/operators/python.py", line 1278 in execute_callable

File "/home/airflow/.local/lib/python3.14/site-packages/airflow/providers/standard/operators/python.py", line 642 in _execute_python_callable_in_subprocess

File "/home/airflow/.local/lib/python3.14/site-packages/airflow/providers/standard/operators/python.py", line 591 in _write_args

Providers standard version is 1.15.0

[Snaptonis]

I figured it out.
I checked the op_args for the task using the expanded task result and in fact they were "LazyXComSequence(<unevaluated length>)" which cannot be serialized.

So I added a normal @task() that takes that XCom and returns it with list(xcom) and pass that to the next task.
Now it works.

Thanks everyone!