Source code for scrapy.utils.trackref

"""This module provides some functions and classes to record and report
references to live object instances.

If you want live objects for a particular class to be tracked, you only have to
subclass from object_ref (instead of object).

About performance: This library has a minimal performance impact when enabled,
and no performance penalty at all when disabled (as object_ref becomes just an
alias to object in that case).

.. note:: PyPy uses a tracing garbage collector, so objects may
    remain in the ``live_refs`` longer than expected, even after they
    go out of scope. If deterministic behavior is required, you may need
    to explicitly trigger garbage collection or call ``trackref.live_refs.clear()``.
"""

from __future__ import annotations

from collections import defaultdict
from operator import itemgetter
from time import monotonic_ns
from types import NoneType
from typing import TYPE_CHECKING, Any
from weakref import WeakKeyDictionary

if TYPE_CHECKING:
    from collections.abc import Iterable

    # typing.Self requires Python 3.11
    from typing_extensions import Self


live_refs: defaultdict[type, WeakKeyDictionary] = defaultdict(WeakKeyDictionary)


[docs] class object_ref: """Inherit from this class to a keep a record of live instances""" __slots__ = () def __new__(cls, *args: Any, **kwargs: Any) -> Self: obj = object.__new__(cls) live_refs[cls][obj] = monotonic_ns() return obj
# using Any as it's hard to type type(None) def format_live_refs(ignore: Any = NoneType) -> str: """Return a tabular representation of tracked objects""" s = "Live References\n\n" now_ns = monotonic_ns() for cls, wdict in sorted(live_refs.items(), key=lambda x: x[0].__name__): if not wdict: continue if issubclass(cls, ignore): continue oldest_ns = min(wdict.values()) s += f"{cls.__name__:<30} {len(wdict):6} oldest: {int((now_ns - oldest_ns) // 1e9)}s ago\n" return s
[docs] def get_oldest(class_name: str) -> Any: """Get the oldest object for a specific class name""" for cls, wdict in live_refs.items(): if cls.__name__ == class_name: if not wdict: break return min(wdict.items(), key=itemgetter(1))[0] return None
[docs] def iter_all(class_name: str) -> Iterable[Any]: """Iterate over all objects of the same class by its class name""" for cls, wdict in live_refs.items(): if cls.__name__ == class_name: return wdict.keys() return ()