Commit 785a1b44 authored by Xavier Barbosa's avatar Xavier Barbosa

current_services() helper

parent a93e4f2b
......@@ -120,6 +120,15 @@ Singleton mode can be disabled per service::
assert result1 != result2
Current services are automatically exposed inside functions::
def func():
return current_services()
assert func() is None
assert (await services.apply(func)) is services
Implementation
--------------
......
from .bases import Injector, annotate
from .bases import Injector, annotate, current_services
from ._version import get_versions
__all__ = ['Injector', 'annotate']
__version__ = get_versions()['version']
__all__ = ["Injector", "annotate"]
__version__ = get_versions()["version"]
del get_versions
from ._version import get_versions
__version__ = get_versions()['version']
__version__ = get_versions()["version"]
del get_versions
......@@ -5,15 +5,27 @@ import concurrent.futures
import logging
from abc import ABCMeta
from collections import OrderedDict, defaultdict
from contextvars import ContextVar
from functools import wraps
from inspect import signature
from itertools import chain
from typing import Optional
from weakref import WeakKeyDictionary
from cached_property import cached_property
logger = logging.getLogger("knighted")
MaybeInjector = Optional["Injector"]
current_services_var: ContextVar[MaybeInjector] = ContextVar(
"current_services", default=None
)
def current_services() -> MaybeInjector:
return current_services_var.get()
class Factory:
def __init__(self, target):
......@@ -151,7 +163,11 @@ class Injector(metaclass=ABCMeta):
result = await result
return result
logger.warning("%r is not annoted", func)
return func(*args, **kwargs)
try:
token = current_services_var.set(self)
return func(*args, **kwargs)
finally:
current_services_var.reset(token)
return wrapper
......
import pytest
from knighted import Injector, annotate
from knighted import Injector, annotate, current_services
import asyncio
from contextlib import contextmanager
from time import time_ns, sleep
......@@ -320,3 +320,11 @@ async def test_not_singleton(services):
sleep(0.1)
result2 = await services.get("foo")
assert result1 != result2
async def test_service_context(services):
def func():
return current_services()
assert func() is None
assert (await services.apply(func)) is services
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment