just depends

depends.py

import asyncio
import inspect


class Depends:
    _cache = {}

    def __init__(self, dependency):
        self._dependency = dependency
        self._resolved = None

    def _resolve(self):
        if self._resolved is not None:
            return self._resolved

        key = self._dependency
        if key in self._cache:
            self._resolved = self._cache[key]
            return self._resolved

        sig = inspect.signature(self._dependency)
        kwargs = {}
        for name, param in sig.parameters.items():
            if isinstance(param.default, Depends):
                kwargs[name] = param.default._resolve()

        result = self._dependency(**kwargs)
        if inspect.iscoroutinefunction(self._dependency):
            try:
                loop = asyncio.get_running_loop()
                import concurrent.futures
                with concurrent.futures.ThreadPoolExecutor() as pool:
                    result = pool.submit(asyncio.run, result).result()
            except RuntimeError:
                result = asyncio.run(result)

        self._cache[key] = result
        self._resolved = result
        return result

    def __getattr__(self, name):
        if name.startswith('_'):
            raise AttributeError(name)
        return getattr(self._resolve(), name)

    def __getitem__(self, key):
        return self._resolve()[key]

    def __repr__(self):
        return repr(self._resolve())

    @classmethod
    def clear(cls):
        cls._cache.clear()


def inject(func):
    def wrapper(*args, **kwargs):
        sig = inspect.signature(func)
        for name, param in sig.parameters.items():
            if isinstance(param.default, Depends) and name not in kwargs:
                kwargs[name] = param.default._resolve()
        result = func(*args, **kwargs)
        if inspect.iscoroutinefunction(func):
            try:
                asyncio.get_running_loop()
                return result  # let caller await
            except RuntimeError:
                return asyncio.run(result)
        return result
    return wrapper


if __name__ == "__main__":

    def get_config():
        print("Loading config...")
        return {"db_url": "postgres://localhost/db"}

    async def get_db(config=Depends(get_config)):
        print(f"Connecting to {config['db_url']}")
        await asyncio.sleep(0.1)
        return {"query": lambda sql: f"Result: {sql}"}

    @inject
    def get_user(db=Depends(get_db)):
        print("Fetching user")
        return {"id": 1, "name": "Alice"}

    print(get_user())