Use methodtools.lru_cache instead of functools.lru_cache in class methods

[Taragolis]

This is attempt to prevent use functools.lru_cache or functools.cache on methods which lead to memory leak, e.g. object can't be garbage collected if it use methods decorated by this decorators.

Simple reproducible example

import functools

class A:
    def __init__(self):
        self.a = {"a": 1, "b": 2}

    @functools.lru_cache(maxsize=None)
    def get_val(self, key: str):
        return self.a[key]

for _ in range(100):
    xxx = A()
    xxx.get_val(key="a")
    xxx.get_val("a")
    xxx.get_val("b")
    del xxx

import gc
gc.collect()
a_instances = [o for o in gc.get_objects() if isinstance(o, A)]
assert len(a_instances) == 0, len(a_instances)
# AssertionError: 100

Instead of that better to use use functools.cached_property (if it could be a property), ruff B018 rule suggestion or use some third-party library, e.g. methodtools

import methodtools

class A:
    def __init__(self):
        self.a = {"a": 1, "b": 2}

    @methodtools.lru_cache(maxsize=None)
    def get_val(self, key: str):
        return self.a[key]

for _ in range(100):
    xxx = A()
    xxx.get_val(key="a")
    xxx.get_val(key="a")
    xxx.get_val("b")
    del xxx

import gc
gc.collect()
a_instances = [o for o in gc.get_objects() if isinstance(o, A)]
assert len(a_instances) == 0, len(a_instances) # Do not raise assertion error

• ruff do not detect compat decorator airflow.compat.functools.cache for avoid use it in methods use simple pre-commit check. It is only required until we have Python 3.9 as min version, and do not use airflow.compat.functools.cache in code base.
• There is couple of places where cache used into the Airflow core, some of them might be a result of memory leakage, the simple way is use methodtools.lru_cache - this one would not breaks public interfaces or behaviour and do no required to rewrite. In another word we have bug fix "Here and Now" instead of think how to rewrite without additional dependency.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[Taragolis]

@potiuk I've also found that ruff do not detect compat decorator airflow.compat.functools.cache and we use it in some methods places into the core, which might cause memory leakage

airflow/airflow/models/abstractoperator.py

Lines 494 to 497 in 8c44bcb

	@cache
	def get_parse_time_mapped_ti_count(self) -> int:
	"""

airflow/airflow/models/mappedoperator.py

Lines 336 to 341 in 8c44bcb

	@classmethod
	@cache
	def get_serialized_fields(cls):
	# Not using 'cls' here since we only want to serialize base fields.
	return frozenset(attr.fields_dict(MappedOperator)) - {

airflow/airflow/models/taskinstance.py

Lines 641 to 643 in fce3a58

	@cache # Prevent multiple database access.
	def _get_previous_dagrun_success() -> DagRun | None:
	return task_instance.get_previous_dagrun(state=DagRunState.SUCCESS, session=session)

airflow/airflow/models/taskinstance.py

Lines 674 to 678 in fce3a58

	@cache
	def get_yesterday_ds() -> str:
	return (logical_date - timedelta(1)).strftime("%Y-%m-%d")

airflow/airflow/models/taskinstance.py

Lines 681 to 685 in fce3a58

	@cache
	def get_tomorrow_ds() -> str:
	return (logical_date + timedelta(1)).strftime("%Y-%m-%d")

airflow/airflow/models/taskinstance.py

Lines 715 to 718 in fce3a58

	@cache
	def get_prev_execution_date():
	# For manually triggered dagruns that aren't run on a schedule,

airflow/airflow/models/taskinstance.py

Lines 727 to 730 in fce3a58

	@cache
	def get_prev_ds() -> str | None:
	execution_date = get_prev_execution_date()

airflow/airflow/utils/task_group.py

Lines 588 to 592 in 0a74928

	@cache
	def get_parse_time_mapped_ti_count(self) -> int:
	"""
	Return the Number of instances a task in this group should be mapped to, when a DAG run is created.

airflow/airflow/utils/weight_rule.py

Lines 37 to 41 in f60d458

	@classmethod
	@cache
	def all_weight_rules(cls) -> set[str]:
	"""Return all weight rules."""
	return set(cls.__members__.values())

airflow/airflow/www/views.py

Lines 336 to 339 in 8c44bcb

	@cache
	def get_task_group_children_getter() -> operator.methodcaller:
	sort_order = conf.get("webserver", "grid_view_sorting_order", fallback="topological")

[Taragolis]

WeightRule is enum, and members of enum are singletons, so it is pretty safe to use it there, but in the other cases it still looks like a problem

[Taragolis]

The problem here with tests that Prod images build do not reflect changes into requirements in hatch_build.py and it doesn't install methodtools

[potiuk]

The problem here with tests that Prod images build do not reflect changes into requirements in hatch_build.py and it doesn't install methodtools

That's already fixed in #38678 -> I rebased your PR @Taragolis

[potiuk]

(and even better - extraction of the common code added then is in #38682 - ready for review.

[Taragolis]

That's already fixed in #38678 -> I rebased your PR @Taragolis

Hehe, I've also rebased it. DOUBLE REBASE 🤣

[Taragolis]

But this failure also show that docker compose tests can't show useful information in case if API failed because it just fail before show logs from the container. I will add workaround for that later on in the separate PR

[potiuk]

But this failure also show that docker compose tests can't show useful information in case if API failed because it just fail before show logs from the container. I will add workaround for that later on in the separate PR

Yep we could dump and upload logs as artifacts from all conainers as we do in case of failure in k8s/regular tests