我们从Python开源项目中，提取了以下26个代码示例，用于说明如何使用sqlalchemy.pool._refs()。
def _assert_no_stray_pool_connections(): global _STRAY_CONNECTION_FAILURES # lazy gc on cPython means "do nothing." pool connections # shouldn't be in cycles, should go away. testutil.lazy_gc() # however, once in awhile, on an EC2 machine usually, # there's a ref in there. usually just one. if pool._refs: # OK, let's be somewhat forgiving. _STRAY_CONNECTION_FAILURES += 1 print("Encountered a stray connection in test cleanup: %s" % str(pool._refs)) # then do a real GC sweep. We shouldn't even be here # so a single sweep should really be doing it, otherwise # there's probably a real unreachable cycle somewhere. testutil.gc_collect() # if we've already had two of these occurrences, or # after a hard gc sweep we still have pool._refs?! # now we have to raise. if pool._refs: err = str(pool._refs) # but clean out the pool refs collection directly, # reset the counter, # so the error doesn't at least keep happening. pool._refs.clear() _STRAY_CONNECTION_FAILURES = 0 assert False, "Stray connection refused to leave "\ "after gc.collect(): %s" % err elif _STRAY_CONNECTION_FAILURES > 10: assert False, "Encountered more than 10 stray connections" _STRAY_CONNECTION_FAILURES = 0
def _assert_no_stray_pool_connections(): global _STRAY_CONNECTION_FAILURES # lazy gc on cPython means "do nothing." pool connections # shouldn't be in cycles, should go away. testutil.lazy_gc() # however, once in awhile, on an EC2 machine usually, # there's a ref in there. usually just one. if pool._refs: # OK, let's be somewhat forgiving. _STRAY_CONNECTION_FAILURES += 1 print("Encountered a stray connection in test cleanup: %s" % str(pool._refs)) # then do a real GC sweep. We shouldn't even be here # so a single sweep should really be doing it, otherwise # there's probably a real unreachable cycle somewhere. testutil.gc_collect() # if we've already had two of these occurrences, or # after a hard gc sweep we still have pool._refs?! # now we have to raise. if pool._refs: err = str(pool._refs) # but clean out the pool refs collection directly, # reset the counter, # so the error doesn't at least keep happening. pool._refs.clear() _STRAY_CONNECTION_FAILURES = 0 warnings.warn( "Stray connection refused to leave " "after gc.collect(): %s" % err) elif _STRAY_CONNECTION_FAILURES > 10: assert False, "Encountered more than 10 stray connections" _STRAY_CONNECTION_FAILURES = 0
def global_cleanup_assertions(): """Check things that have to be finalized at the end of a test suite. Hardcoded at the moment, a modular system can be built here to support things like PG prepared transactions, tables all dropped, etc. """ testutil.lazy_gc() assert not pool._refs, str(pool._refs)
