How to use the asgiref.local.Local function in asgiref

gc.collect()

        # Trigger the local creation outside
        e1.set()
        e2.wait()

        # New Locals should be empty
        matched = len(
            [local for local in locals if getattr(local, "foo", None) == "bar"]
        )

    t = threading.Thread(target=f)
    t.start()
    e1.wait()
    # Creates locals outside of the inner thread
    locals = [Local() for i in range(100)]
    e2.set()
    t.join()

    assert matched == 0

def test_local_critical_no_thread_to_task():
    """
    Tests that local does not go through async_to_sync when the local is set
    as thread-critical
    """
    # Set up the local
    test_local = Local(thread_critical=True)
    test_local.foo = 89
    # Look at it in an async context
    async def async_function():
        with pytest.raises(AttributeError):
            test_local.foo
        test_local.foo = "numbers"
        assert test_local.foo == "numbers"

    async_to_sync(async_function)()
    # Check the value outside
    assert test_local.foo == 89

def test_pgettext(self):
        trans_real._active = Local()
        trans_real._translations = {}
        with translation.override('de'):
            self.assertEqual(pgettext("unexisting", "May"), "May")
            self.assertEqual(pgettext("month name", "May"), "Mai")
            self.assertEqual(pgettext("verb", "May"), "Kann")
            self.assertEqual(npgettext("search", "%d result", "%d results", 4) % 4, "4 Resultate")

def test_local_thread_to_async():
    """
    Tests that local carries through async_to_sync
    """
    # Set up the local
    test_local = Local()
    test_local.foo = 12
    # Look at it in an async context
    async def async_function():
        assert test_local.foo == 12
        test_local.foo = "inside"

    async_to_sync(async_function)()
    # Check the value passed out again
    assert test_local.foo == "inside"

def test_local_del_swallows_type_error(monkeypatch):
    test_local = Local()

    blow_up_calls = 0

    def blow_up(self):
        nonlocal blow_up_calls
        blow_up_calls += 1
        raise TypeError()

    monkeypatch.setattr("weakref.WeakSet.__iter__", blow_up)

    test_local.__del__()

    assert blow_up_calls == 1

def test_safe_status(self):
        """
        Translating a string requiring no auto-escaping with gettext or pgettext
        shouldn't change the "safe" status.
        """
        trans_real._active = Local()
        trans_real._translations = {}
        s1 = mark_safe('Password')
        s2 = mark_safe('May')
        with translation.override('de', deactivate=True):
            self.assertIs(type(gettext(s1)), SafeString)
            self.assertIs(type(pgettext('month name', s2)), SafeString)
        self.assertEqual('aPassword', SafeString('a') + s1)
        self.assertEqual('Passworda', s1 + SafeString('a'))
        self.assertEqual('Passworda', s1 + mark_safe('a'))
        self.assertEqual('aPassword', mark_safe('a') + s1)
        self.assertEqual('as', mark_safe('a') + mark_safe('s'))

def test_local_thread():
    """
    Tests that local works just inside a normal thread context
    """

    test_local = Local()
    # Unassigned should be an error
    with pytest.raises(AttributeError):
        test_local.foo
    # Assign and check it persists
    test_local.foo = 2
    assert test_local.foo == 2
    # Delete and check it errors again
    del test_local.foo
    with pytest.raises(AttributeError):
        test_local.foo

def __init__(self, databases=None):
        """
        databases is an optional dictionary of database definitions (structured
        like settings.DATABASES).
        """
        self._databases = databases
        # Connections needs to still be an actual thread local, as it's truly
        # thread-critical. Database backends should use @async_unsafe to protect
        # their code from async contexts, but this will give those contexts
        # separate connections in case it's needed as well. There's no cleanup
        # after async contexts, though, so we don't allow that if we can help it.
        self._connections = Local(thread_critical=True)

from django.utils.encoding import iri_to_uri
from django.utils.functional import lazy
from django.utils.translation import override

from .exceptions import NoReverseMatch, Resolver404
from .resolvers import _get_cached_resolver, get_ns_resolver, get_resolver
from .utils import get_callable

# SCRIPT_NAME prefixes for each thread are stored here. If there's no entry for
# the current thread (which is the only one we ever access), it is assumed to
# be empty.
_prefixes = Local()

# Overridden URLconfs for each thread are stored here.
_urlconfs = Local()


def resolve(path, urlconf=None):
    if urlconf is None:
        urlconf = get_urlconf()
    return get_resolver(urlconf).resolve(path)


def reverse(viewname, urlconf=None, args=None, kwargs=None, current_app=None):
    if urlconf is None:
        urlconf = get_urlconf()
    resolver = get_resolver(urlconf)
    args = args or []
    kwargs = kwargs or {}

    prefix = get_script_prefix()

def translation_file_changed(sender, file_path, **kwargs):
    """Clear the internal translations cache if a .mo file is modified."""
    if file_path.suffix == '.mo':
        import gettext
        from django.utils.translation import trans_real
        gettext._translations = {}
        trans_real._translations = {}
        trans_real._default = None
        trans_real._active = Local()
        return True