Override CustomField dependency

[chrisgoddard]

Hi there - how are you supposed to override a custom field dependency?

In the canonical example:

from fast_depends import inject
from fast_depends.library import CustomField

class Header(CustomField):
    def use(self, /, **kwargs: AnyDict) -> AnyDict:
        kwargs = super().use(**kwargs)
        kwargs[self.param_name] = kwargs["headers"][self.param_name]
        return kwargs

@inject
def my_func(header_field: int = Header()):
    return header_field

assert my_func(
    headers={ "header_field": "1" }
) == 1

If we wanted to override the Header() dependency using dependency_provider - how would we do that? I'm not sure how to access the call property of the CustomField instance.

e.g.

from fast_depends import dependency_provider

def mock_provider() -> MockHeader:
    pass

@inject
def my_func(header_field: int = Header()):
    return header_field


dependency_provider.dependency_overrides[WHAT_GOES_HERE] = mock_provider

assert my_func(
    headers={ "header_field": "1" }
) == 1

[Lancetnik]

@chrisgoddard, sorry, overrides are only working for Depends at the moment. I'll see what we can do about it after the 3.0.0 release.

[chrisgoddard]

Got it - do you have a timeline on 3.0 release?

[chrisgoddard]

I use this lib a lot so let me know if you need any help with 3.0 development

[Lancetnik]

I use this lib a lot so let me know if you need any help with 3.0 development

Thank you! It is now almost stable. You can even install it using fast-depends==3.0.0a5.

I am currently migrating FastStream 0.6 to this version and releasing both libraries. It would be great if you could try upgrading your current project to version 3.0.0 and report any bugs that occur, so I can fix them before the stable release.

[Lancetnik]

@chrisgoddard, also, thank you for being with all my projects throughout this time! It looks like you are the oldest Propan / FastStream and FastDepends user! 😄

May I ask you: Are you still using FastStream in your company, or is it just FastDepends now?
I would like to improve our landing page by adding publicly verified company users, and your feedback would be greatly appreciated.

[chrisgoddard]

I'll try to upgrade and test today!

Wow - that's crazy! Thanks for all the work you've put into it.

So I'm not actively using FastStream at the moment because I moved my focus from a streaming-focused system to a more traditional task/workflow orchestration approach - but it is still my go-to when I do have streaming applications and I'm hoping to reintroduce some streaming functionality into one of my projects soon!

[chrisgoddard]

Do you have any docs about the changes in v3? I'm trying to test it out but I think part of the problem is that I built some of my own code on top of fast depends in a way where I make certain assumptions about fast_depend's internals which have now changed. In particular, can you explain the cast vs cast_result and the dependency_provider parameter in Depends?

[Lancetnik]

Do you have any docs about the changes in v3? I'm trying to test it out but I think part of the problem is that I built some of my own code on top of fast depends in a way where I make certain assumptions about fast_depend's internals which have now changed. In particular, can you explain the cast vs cast_result and the dependency_provider parameter in Depends?

Sorry, I tried to save public API, but inner realization changed a lot...
cast is an option controls types casting at all. If exists for backport compatibility-only. In fact it is just a shortcut to serializer_cls=None
cast_result is a serailizer option controls func() -> result type casting. So, you can enable casting for incoming options and disable for outgoing
dependency_provider is a same object for overrides, but I tried to remove the global one. Now you can pass your own dependency_provider object to @inject instead of using default global object.

[chrisgoddard]

Hey there - just confirming I've been able to update my application to use the fast-depends v3 with no issues

[Lancetnik]

Hey there - just confirming I've been able to update my application to use the fast-depends v3 with no issues

Thank you! It's really great news for me!

[chrisgoddard]

Sorry - I realized that I was so focused on testing the new version for compatibility with my existing code base I never checked the original use case for my question.

If I have the example I gave above:

from fast_depends import inject
from fast_depends.library import CustomField

class Header(CustomField):
    def use(self, /, **kwargs: AnyDict) -> AnyDict:
        kwargs = super().use(**kwargs)
        kwargs[self.param_name] = kwargs["headers"][self.param_name]
        return kwargs

@inject
def my_func(header_field: int = Header()):
    return header_field

assert my_func(
    headers={ "header_field": "1" }
) == 1

In version 3.X, how do I override the header_field dependency at runtime for testing purposes?

I see the custom provider logic but I'm not 100% sure how to modify that for a given function at runtime

[Lancetnik]

@chrisgoddard, I'm sorry, but there is no such option available yet. I haven't designed CustomFields to be overridable at this time. However, I will try to come up with something.

By the way, can you tell me more about your use case for CustomFields? I am really interested in hearing about it.

[chrisgoddard]

Ah ok - yeah please keep me posted - I make fairly extensive use of them.

Essentially I've standardized all of my api clients to use extend BaseProvider class:

import copy
import typing
import asyncio
import warnings
from fast_depends.library import CustomField
from loguru import logger
import inspect

T = typing.TypeVar("T")


def _get_generic_type(cls):
    """Extract the generic type parameter from BaseProvider[T] or AsyncBaseProvider[T]"""
    # Check if __orig_bases__ exists (Python 3.7+)
    if hasattr(cls, '__orig_bases__'):
        for base in cls.__orig_bases__:
            origin = typing.get_origin(base)
            # Check if this is our generic base class
            if origin is BaseProvider or (origin and issubclass(origin, BaseProvider)):
                args = typing.get_args(base)
                if args:
                    return args[0]
            elif origin is AsyncBaseProvider or (origin and issubclass(origin, AsyncBaseProvider)):
                args = typing.get_args(base)
                if args:
                    return args[0]
    
    # Fallback to old behavior for backward compatibility
    # This handles the case where the class inherits from both BaseProvider[T] and T
    if len(cls.__bases__) > 1:
        for base in cls.__bases__:
            # Skip the provider base classes
            if not (base is BaseProvider or base is AsyncBaseProvider or 
                    (hasattr(base, '__origin__') and typing.get_origin(base) in (BaseProvider, AsyncBaseProvider))):
                return base
    
    return None


class BaseProvider(CustomField, typing.Generic[T]):
    __override_class__: typing.ClassVar = None

    @classmethod
    def provide(cls, *args, **kwargs) -> T:
        _kwargs = kwargs.copy()
        _target_cls = cls.__override_class__ or _get_generic_type(cls)
        if _target_cls is None:
            raise TypeError(f"Could not determine target class for {cls.__name__}. "
                          "Make sure to inherit from BaseProvider[YourClass].")

        _args = inspect.getfullargspec(_target_cls.__init__)

        if not _args.varkw:
            _remove_args = set()
            for k, v in kwargs.items():
                if k not in _args.args:
                    # logger.debug(f"Removing {k}")
                    _remove_args.add(k)
                    _kwargs.pop(k)

            if _remove_args:
                logger.debug(f"Removing {len(_remove_args)} args: {_remove_args}")

        kwargs = _kwargs

        if cls.__override_class__ is not None:
            return cls.__override_class__(*args, **kwargs)
        else:
            return _target_cls(*args, **kwargs)

    def __init__(self, *args, _debug: bool = False, **kwargs):
        super().__init__(cast=False)
        self._args = args
        self._kwargs = kwargs
        self._debug = _debug
        self._warning_shown = False

    def initializer(
        self,
        cls_args: tuple[typing.Any, ...],
        cls_kwargs: dict[str, typing.Any],
        fn_kwargs: dict[str, typing.Any],
    ):
        return cls_args, (cls_kwargs or {}) | (fn_kwargs or {})

    def use(self, /, **kwargs: dict[str, typing.Any]):
        if self._debug:
            logger.debug(f"Using {self.__class__.__name__}: {kwargs}")
        
        # Check if using old pattern and show deprecation warning
        if not self._warning_shown and self.param_name and self._is_used_as_default():
            self._show_deprecation_warning()
            self._warning_shown = True
        
        kwargs = super().use(**kwargs)
        if self.param_name:
            _args, _kwargs = self.initializer(
                cls_args=copy.copy(self._args),
                cls_kwargs=copy.copy(self._kwargs),
                fn_kwargs=kwargs,
            )
            kwargs[self.param_name] = self.provide(*_args, **_kwargs)  # type: ignore
        return kwargs

    def get(self, **kwargs) -> T:
        self.param_name = "_default"
        return self.use(**kwargs).get(self.param_name)  # type: ignore
    
    def _is_used_as_default(self) -> bool:
        """Check if this provider is being used as a function default value"""
        # Inspect the call stack to find the injected function
        frame = inspect.currentframe()
        try:
            # Walk up the stack to find the decorated function
            while frame:
                frame_locals = frame.f_locals
                
                # Look for function objects that might contain this provider as a default
                for name, obj in frame_locals.items():
                    if callable(obj):
                        # Check both the function and its wrapped version (for @inject)
                        funcs_to_check = [obj]
                        if hasattr(obj, '__wrapped__'):
                            funcs_to_check.append(obj.__wrapped__)
                        
                        for func in funcs_to_check:
                            if hasattr(func, '__defaults__') and func.__defaults__:
                                if self in func.__defaults__:
                                    # Check if this function parameter uses Annotated
                                    sig = inspect.signature(func)
                                    for param_name, param in sig.parameters.items():
                                        if param.default is self:
                                            # Check if the annotation is Annotated
                                            if typing.get_origin(param.annotation) is not typing.Annotated:
                                                return True
                                    
                frame = frame.f_back
        finally:
            del frame
        
        return False
    
    def _show_deprecation_warning(self):
        """Show deprecation warning with migration guide"""
        target_type = _get_generic_type(self.__class__)
        type_name = target_type.__name__ if target_type else "YourType"
        
        warnings.warn(
            f"Using {self.__class__.__name__} as a default parameter value is deprecated and will be removed in a future version.\n"
            f"Please use the Annotated pattern instead:\n"
            f"  Old: func(param: {type_name} = {self.__class__.__name__}(...))\n"
            f"  New: func(param: Annotated[{type_name}, {self.__class__.__name__}(...)])",
            DeprecationWarning,
            stacklevel=6  # Adjusted to show the user's code
        )


class AsyncBaseProvider(CustomField, typing.Generic[T]):
    __override_class__: typing.ClassVar = None

    async def get(self, **kwargs) -> T:
        self.param_name = "_default"
        return (await self.use(**kwargs)).get(self.param_name)

    @classmethod
    async def provide(cls, *args, **kwargs) -> T:
        if cls.__override_class__ is not None:
            return cls.__override_class__(*args, **kwargs)
        else:
            _target_cls = _get_generic_type(cls)
            if _target_cls is None:
                raise TypeError(f"Could not determine target class for {cls.__name__}. "
                              "Make sure to inherit from AsyncBaseProvider[YourClass].")
            return _target_cls(*args, **kwargs)

    def __init__(self, *args, _debug: bool = False, **kwargs):
        super().__init__(cast=False)
        self._args = args
        self._kwargs = kwargs
        self._debug = _debug
        self._warning_shown = False

    def initializer(
        self,
        cls_args: tuple[typing.Any, ...],
        cls_kwargs: dict[str, typing.Any],
        fn_kwargs: dict[str, typing.Any],
    ):
        return cls_args, (cls_kwargs or {}) | (fn_kwargs or {})

    async def use(self, /, **kwargs: dict[str, typing.Any]):
        if self._debug:
            logger.debug(f"Using {self.__class__.__name__}: {kwargs}")
        
        # Check if using old pattern and show deprecation warning
        if not self._warning_shown and self.param_name and self._is_used_as_default():
            self._show_deprecation_warning()
            self._warning_shown = True
            
        kwargs = super().use(**kwargs)
        if self.param_name:
            _args, _kwargs = self.initializer(
                cls_args=copy.copy(self._args),
                cls_kwargs=copy.copy(self._kwargs),
                fn_kwargs=kwargs,
            )
            kwargs[self.param_name] = await self.provide(*_args, **_kwargs)  # type: ignore

            await self.on_initialize(
                kwargs[self.param_name],
                **{k: v for k, v in _kwargs.items() if k != self.param_name},
            )
        return kwargs

    async def on_initialize(self, instance, **kwargs):
        pass
    
    def _is_used_as_default(self) -> bool:
        """Check if this provider is being used as a function default value"""
        # Inspect the call stack to find the injected function
        frame = inspect.currentframe()
        try:
            # Walk up the stack to find the decorated function
            while frame:
                frame_locals = frame.f_locals
                
                # Look for function objects that might contain this provider as a default
                for name, obj in frame_locals.items():
                    if callable(obj):
                        # Check both the function and its wrapped version (for @inject)
                        funcs_to_check = [obj]
                        if hasattr(obj, '__wrapped__'):
                            funcs_to_check.append(obj.__wrapped__)
                        
                        for func in funcs_to_check:
                            if hasattr(func, '__defaults__') and func.__defaults__:
                                if self in func.__defaults__:
                                    # Check if this function parameter uses Annotated
                                    sig = inspect.signature(func)
                                    for param_name, param in sig.parameters.items():
                                        if param.default is self:
                                            # Check if the annotation is Annotated
                                            if typing.get_origin(param.annotation) is not typing.Annotated:
                                                return True
                                    
                frame = frame.f_back
        finally:
            del frame
        
        return False
    
    def _show_deprecation_warning(self):
        """Show deprecation warning with migration guide"""
        target_type = _get_generic_type(self.__class__)
        type_name = target_type.__name__ if target_type else "YourType"
        
        warnings.warn(
            f"Using {self.__class__.__name__} as a default parameter value is deprecated and will be removed in a future version.\n"
            f"Please use the Annotated pattern instead:\n"
            f"  Old: func(param: {type_name} = {self.__class__.__name__}(...))\n"
            f"  New: func(param: Annotated[{type_name}, {self.__class__.__name__}(...)])",
            DeprecationWarning,
            stacklevel=6  # Adjusted to show the user's code
        )

I can then customize how it is initialized - for instance reading parameters from environment variables, etc

For instance, this ClickhouseProvider:

class ClickhouseProvider(BaseProvider[Clickhouse]):
    def __init__(
        self,
        profile: str | None = None,
        host: str | None = None,
        port: int | None = None,
        database: str | None = None,
        user: str | None = None,
        password: str | None = None,
        secure: bool | None = None,
        compression: bool | None = None,
        # _debug: bool = False,
        wake: bool = False,
    ):
        super().__init__(
            profile=profile,
            host=host,
            port=port,
            database=database,
            user=user,
            password=password,
            secure=secure,
            compression=compression,
            wake=wake,
        )

    def initializer(self, cls_args, cls_kwargs, fn_kwargs):
        # mode = {**cls_kwargs, **fn_kwargs}.get("profile", "cloud").upper()

        _allowable_args = {
            "dsn",
            "host",
            "port",
            "database",
            "user",
            "password",
            "secure",
            "compression",
        }

        kwargs = {**cls_kwargs, **fn_kwargs}

        profile_name = kwargs.pop("profile", None)

        if profile_name:
            profile = (
                "CLICKHOUSE_" + profile_name.upper()
                if "clickhouse" not in profile_name.lower()
                else profile_name
            )
            _overwrite_settings = {}
            for arg in kwargs.keys():
                if (
                    arg not in ("profile", "host", "port", "user", "password")
                    and kwargs.get(arg) is not None
                ):
                    _overwrite_settings[arg] = kwargs[arg]

            profile_args = ConnectionProfile.load_by_prefix(
                profile, os.environ
            ).model_dump(
                mode="json",
                exclude_none=True,
            )

            kwargs = {
                **profile_args,
                **_overwrite_settings,
            }
        # logger.info({k for k,v in kwargs.items() if k not in _allowable_args})
        kwargs = {
            k: v for k, v in kwargs.items() if k in _allowable_args and v is not None
        }

        return cls_args, kwargs

    @classmethod
    def provide(cls, *args, **kwargs) -> Clickhouse:
        wake = kwargs.pop("wake", False)
        return Clickhouse(connection=connect(**kwargs), wake=wake)

Which I can then use via:

@inject
def some_function(ch: Clickhouse = ClickhouseProvider()):
    ...

It's admittedly been a while since I revisited this code - but I've been relying on it ever since I started using fast depends and it's been my go-to. I'm sure it could be cleaned up/improved upon and maybe there's even something I could do in this implementation to facilitate overrides myself