#16037 Templated requirements.txt in Python operators

[rounakdatta]

Closes: #16037
This pull request addresses the changes discussed in the aforementioned issue.

• Added new unit tests
• This change is not breaking
• Updating documentation of the operator

Changes:

• Support for passing a requirements.txt file in the requirements field of the PythonVirtualenvOperator.
• The file can be named *.txt (anything ending with .txt), and can be jinja templated.

Note: template_searchpath must be appropriately set while using arbitrary locations of the template file.

[github-actions]

The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.

[potiuk]

I think we need to support "prepare_virtualenv" backwards compatible behaviour.

[potiuk]

I think we should still (backwards compatibility) handle the case where requirements are List[str]. It will make it a bit more complex, but I think it is needed.

I propose to keep the old requrements handling only List[str] and add a new parameter requirements_file_path - and check if only one of those is passed (and act accordingly). I think it is very ambiguous to name requirements something that is path to requirements file.

[rounakdatta]

Addressed! 😃

[uranusjr]

Come on CI

[rounakdatta]

Looks like one of the tests is stuck, shall I push an empty commit @uranusjr ?

[uranusjr]

No need, I just re-triggered them.

[rounakdatta]

One check is failing where apache-airflow[devel-ci] package is failing to install because of a dependency version incompatibility. I believe this didn't get introduced due to my change 😬

[potiuk]

One check is failing where apache-airflow[devel-ci] package is failing to install because of a dependency version incompatibility. I believe this didn't get introduced due to my change 😬

I think you need to rebase - that was celery-5 change that got merged recently, and it is conflcting unless you rebase to latest main.

[kaxil]

tests are failing:

=================================== FAILURES ===================================
  __________________ TestPythonVirtualenvOperator.test_add_dill __________________
  
  self = <tests.operators.test_python.TestPythonVirtualenvOperator testMethod=test_add_dill>
  
      def test_add_dill(self):
          def f():
              import dill  # noqa: F401
              import lazy_object_proxy  # noqa: F401
      
  >       self._run_as_operator(f, use_dill=True, system_site_packages=False)
  
  tests/operators/test_python.py:750: 
  _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  tests/operators/test_python.py:742: in _run_as_operator
      task.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
  airflow/utils/session.py:69: in wrapper
      return func(*args, session=session, **kwargs)
  airflow/models/baseoperator.py:1348: in run
      session=session,
  airflow/utils/session.py:66: in wrapper
      return func(*args, **kwargs)
  airflow/models/taskinstance.py:1636: in run
      session=session,
  airflow/utils/session.py:66: in wrapper
      return func(*args, **kwargs)
  airflow/models/taskinstance.py:1331: in _run_raw_task
      self._execute_task_with_callbacks(context)
  airflow/models/taskinstance.py:1457: in _execute_task_with_callbacks
      result = self._execute_task(context, self.task)
  airflow/models/taskinstance.py:1513: in _execute_task
      result = execute_callable(context=context)
  airflow/operators/python.py:408: in execute
      return super().execute(context=serializable_context)
  airflow/operators/python.py:181: in execute
      return_value = self.execute_callable()
  airflow/operators/python.py:460: in execute_callable
      string_args_filename,

[uranusjr]

Those test_add_dill tests shouldn’t import lazy_object_proxy, that is not guaranteed to be injected.

[uranusjr]

I committed and pushed the previous review suggestions directly to the branch.

[boring-cyborg]

Awesome work, congrats on your first merged pull request!

[basjacobs93]

This does not seem to be working properly when requirements is a string which contains a file path to a requirements file. If I provide requirements with a path to a requirements.txt file, I get (I obfuscated the pathname)

ERROR: Invalid requirement: '<pathname>/requirements.txt' (from line 1 of /tmp/venvbcsbi3im/requirements.txt)
Hint: It looks like a path. The path does exist. The argument you provided (<pathname>/requirements.txt) appears to be a requirements file. If that is the case, use the '-r' flag to install the packages specified within it.

In this line, the requirements parameter (in this case a file path) is stored into self.requirements:

airflow/airflow/operators/python.py

Line 399 in b597cea

self.requirements = requirements

Then, this file path is put into the requirements_file_contents variable

airflow/airflow/operators/python.py

Line 423 in b597cea

requirements_file_contents = self.requirements

Next, this file path (rather than its contents) is written to the f'{tmp_dir}/requirements.txt' file

airflow/airflow/operators/python.py

Line 429 in b597cea

file.write(requirements_file_contents)

Indeed, if I pass it requirements="-r <pathname>/requirements.txt", it works correctly.

I think it would be best to change

airflow/airflow/operators/python.py

Line 399 in b597cea

self.requirements = requirements

to

with open(requirements, "r") as file:
    self.requirements = file.readlines()

[uranusjr]

A pull request would be much appreciated.

[basjacobs93]

See PR #24368
The contents of the file requirements are now stored into self.requirements. These are then saved to the temporary requirements file which is passed to pip.