danswer/backend/onyx/utils/object_size_check.py
Chris Weaver f1fc8ac19b
Connector checkpointing (#3876)
* wip checkpointing/continue on failure

more stuff for checkpointing

Basic implementation

FE stuff

More checkpointing/failure handling

rebase

rebase

initial scaffolding for IT

IT to test checkpointing

Cleanup

cleanup

Fix it

Rebase

Add todo

Fix actions IT

Test more

Pagination + fixes + cleanup

Fix IT networking

fix it

* rebase

* Address misc comments

* Address comments

* Remove unused router

* rebase

* Fix mypy

* Fixes

* fix it

* Fix tests

* Add drop index

* Add retries

* reset lock timeout

* Try hard drop of schema

* Add timeout/retries to downgrade

* rebase

* test

* test

* test

* Close all connections

* test closing idle only

* Fix it

* fix

* try using null pool

* Test

* fix

* rebase

* log

* Fix

* apply null pool

* Fix other test

* Fix quality checks

* Test not using the fixture

* Fix ordering

* fix test

* Change pooling behavior
2025-02-16 02:34:39 +00:00

27 lines
721 B
Python

import sys
from typing import TypeVar
T = TypeVar("T", dict, list, tuple, set, frozenset)
def deep_getsizeof(obj: T, seen: set[int] | None = None) -> int:
"""Recursively sum size of objects, handling circular references."""
if seen is None:
seen = set()
obj_id = id(obj)
if obj_id in seen:
return 0 # Prevent infinite recursion for circular references
seen.add(obj_id)
size = sys.getsizeof(obj)
if isinstance(obj, dict):
size += sum(
deep_getsizeof(k, seen) + deep_getsizeof(v, seen) for k, v in obj.items()
)
elif isinstance(obj, (list, tuple, set, frozenset)):
size += sum(deep_getsizeof(i, seen) for i in obj)
return size