Skip to content

selfdependent()

Decorator that converts an instance method into a FastAPI dependency using a SelfDependent descriptor.

Parameters:

Name Type Description Default
factory Dependency[TOwner] | None

An optional factory (and FastAPI dependency) that can be wrapped in Depends() and that produces the self argument for the wrapped method.

 None
Source code in fasted/selfdependent.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
def selfdependent(
    *,
    factory: Dependency[TOwner] | None = None,
) -> Callable[[BoundMethod[TOwner, TParams, TResult]], BoundMethod[TOwner, TParams, TResult]]:
    """
    Decorator that converts an instance method into a FastAPI dependency using a `SelfDependent` descriptor.

    Arguments:
        factory: An optional factory (and FastAPI dependency) that can be wrapped in `Depends()`
            and that produces the `self` argument for the wrapped method.
    """

    def decorator(
        func: Callable[Concatenate[TOwner, TParams], TResult],
    ) -> SelfDependent[TOwner, TParams, TResult]:
        return SelfDependent[TOwner, TParams, TResult](wrapped=func, factory=factory)

    # This type ignore is necessary for mypy in case the decorated method is overridden in a subclass.
    return decorator  # type: ignore

SelfDependent

Bases: Generic[TOwner, TParams, TResult]

Descriptor whose value is a method (FastAPI dependency) with an annotated self argument that can be processed and used by FastAPI as a dependency.

Source code in fasted/selfdependent.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
class SelfDependent(Generic[TOwner, TParams, TResult]):
    """
    Descriptor whose value is a method (FastAPI dependency) with an annotated
    `self` argument that can be processed and used by FastAPI as a dependency.
    """

    __slots__ = ("_wrapped", "_factory", "_memo")

    def __init__(
        self,
        wrapped: Callable[Concatenate[TOwner, TParams], TResult],
        *,
        factory: Dependency[TOwner] | None = None,
    ) -> None:
        """
        Initialization.

        Arguments:
            wrapped: The wrapped function.
            factory: An optional factory for creating `self` instances.
        """
        self._wrapped = wrapped
        self._factory = factory
        self._memo = IdMemo[Callable[TParams, TResult]]()

    def __get__(self, owner: TOwner | None, obj_type: type[TOwner]) -> Callable[TParams, TResult]:
        memo = self._memo
        hcurrent = memo.hash(owner, obj_type)
        if hcurrent in memo:
            return self._memo.value

        result: Callable[TParams, TResult]
        wrapped = self._wrapped
        if inspect.isgeneratorfunction(wrapped):
            result = SelfWrapper.sync_generator(wrapped, owner)  # type: ignore[assignment]
        elif inspect.isasyncgenfunction(wrapped):
            result = SelfWrapper.async_generator(wrapped, owner)  # type: ignore[assignment]
        elif asyncio.iscoroutinefunction(wrapped):
            result = SelfWrapper.async_method(wrapped, owner)  # type: ignore[assignment]
        else:
            result = SelfWrapper.sync_method(wrapped, owner)

        replace_self_signature(
            result,
            inspect.Parameter(
                "self",
                inspect.Parameter.POSITIONAL_OR_KEYWORD,
                annotation=Annotated[obj_type, Depends(self._factory or obj_type)],
            ),
        )

        return self._memo.store(hcurrent, result)

__init__(wrapped, *, factory=None)

Initialization.

Parameters:

Name Type Description Default
wrapped Callable[Concatenate[TOwner, TParams], TResult]

The wrapped function.

 required
factory Dependency[TOwner] | None

An optional factory for creating self instances.

 None
Source code in fasted/selfdependent.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def __init__(
    self,
    wrapped: Callable[Concatenate[TOwner, TParams], TResult],
    *,
    factory: Dependency[TOwner] | None = None,
) -> None:
    """
    Initialization.

    Arguments:
        wrapped: The wrapped function.
        factory: An optional factory for creating `self` instances.
    """
    self._wrapped = wrapped
    self._factory = factory
    self._memo = IdMemo[Callable[TParams, TResult]]()

SelfWrapper

SelfDependent function wrappers that populate the self argument of the wrapped function from the keyword arguments (if one was provided), or use the owner instance if one wasn't provided. Thus the wrapped "static" methods can behave as if they were instance methods.

Source code in fasted/selfdependent.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class SelfWrapper:
    """
    `SelfDependent` function wrappers that populate the `self` argument of the wrapped function
    from the keyword arguments (if one was provided), or use the owner instance if one wasn't
    provided. Thus the wrapped "static" methods can behave as if they were instance methods.
    """

    @classmethod
    def sync_method(
        cls,
        func: Callable[Concatenate[TOwner, TParams], TResult],
        owner: TOwner | None,
    ) -> Callable[TParams, TResult]:
        """
        Wrapper for synchronous methods.

        Arguments:
            func: The function to wrap.
            owner: An optional owner object.

        Returns:
            The wrapper.
        """

        @wraps(func)
        def do(*args: TParams.args, **kwargs: TParams.kwargs) -> TResult:
            func_self: TOwner = kwargs.pop("self") if "self" in kwargs else owner  # type: ignore[assignment]
            if func_self is None:
                raise RuntimeError("Missing self argument.")

            return func(func_self, *args, **kwargs)

        return do

    @classmethod
    def sync_generator(
        cls,
        func: Callable[Concatenate[TOwner, TParams], Generator[TResult, Any, Any]],
        owner: TOwner | None,
    ) -> Callable[TParams, Generator[TResult, None, None]]:
        """
        Wrapper for synchronous generator methods.

        Arguments:
            func: The function to wrap.
            owner: An optional owner object.

        Returns:
            The wrapper.
        """

        @wraps(func)
        def do(*args: TParams.args, **kwargs: TParams.kwargs) -> Generator[TResult, None, None]:
            func_self: TOwner = kwargs.pop("self") if "self" in kwargs else owner  # type: ignore[assignment]
            if func_self is None:
                raise RuntimeError("Missing self argument.")

            yield from func(func_self, *args, **kwargs)

        return do

    @classmethod
    def async_method(
        cls,
        func: Callable[Concatenate[TOwner, TParams], Coroutine[None, None, TResult]],
        owner: TOwner | None,
    ) -> Callable[TParams, Coroutine[None, None, TResult]]:
        """
        Wrapper for asynchronous methods.

        Arguments:
            func: The function to wrap.
            owner: An optional owner object.

        Returns:
            The wrapper.
        """

        @wraps(func)
        async def do(*args: TParams.args, **kwargs: TParams.kwargs) -> TResult:
            func_self: TOwner = kwargs.pop("self") if "self" in kwargs else owner  # type: ignore[assignment]
            if func_self is None:
                raise RuntimeError("Missing self argument.")

            return await func(func_self, *args, **kwargs)

        return do

    @classmethod
    def async_generator(
        cls,
        func: Callable[
            Concatenate[TOwner, TParams],
            AsyncGenerator[TResult, None],
        ],
        owner: TOwner | None,
    ) -> Callable[TParams, AsyncGenerator[TResult, None]]:
        """
        Wrapper for asynchronous generator methods.

        Arguments:
            func: The function to wrap.
            owner: An optional owner object.

        Returns:
            The wrapper.
        """

        @wraps(func)
        async def do(*args: TParams.args, **kwargs: TParams.kwargs) -> AsyncGenerator[TResult, None]:
            func_self: TOwner = kwargs.pop("self") if "self" in kwargs else owner  # type: ignore[assignment]
            if func_self is None:
                raise RuntimeError("Missing self argument.")

            async for res in func(func_self, *args, **kwargs):
                yield res

        return do

async_generator(func, owner) classmethod

Wrapper for asynchronous generator methods.

Parameters:

Name Type Description Default
func Callable[Concatenate[TOwner, TParams], AsyncGenerator[TResult, None]]

The function to wrap.

 required
owner TOwner | None

An optional owner object.

 required

Returns:

Type Description
Callable[TParams, AsyncGenerator[TResult, None]]

The wrapper.
Source code in fasted/selfdependent.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@classmethod
def async_generator(
    cls,
    func: Callable[
        Concatenate[TOwner, TParams],
        AsyncGenerator[TResult, None],
    ],
    owner: TOwner | None,
) -> Callable[TParams, AsyncGenerator[TResult, None]]:
    """
    Wrapper for asynchronous generator methods.

    Arguments:
        func: The function to wrap.
        owner: An optional owner object.

    Returns:
        The wrapper.
    """

    @wraps(func)
    async def do(*args: TParams.args, **kwargs: TParams.kwargs) -> AsyncGenerator[TResult, None]:
        func_self: TOwner = kwargs.pop("self") if "self" in kwargs else owner  # type: ignore[assignment]
        if func_self is None:
            raise RuntimeError("Missing self argument.")

        async for res in func(func_self, *args, **kwargs):
            yield res

    return do

async_method(func, owner) classmethod

Wrapper for asynchronous methods.

Parameters:

Name Type Description Default
func Callable[Concatenate[TOwner, TParams], Coroutine[None, None, TResult]]

The function to wrap.

 required
owner TOwner | None

An optional owner object.

 required

Returns:

Type Description
Callable[TParams, Coroutine[None, None, TResult]]

The wrapper.
Source code in fasted/selfdependent.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@classmethod
def async_method(
    cls,
    func: Callable[Concatenate[TOwner, TParams], Coroutine[None, None, TResult]],
    owner: TOwner | None,
) -> Callable[TParams, Coroutine[None, None, TResult]]:
    """
    Wrapper for asynchronous methods.

    Arguments:
        func: The function to wrap.
        owner: An optional owner object.

    Returns:
        The wrapper.
    """

    @wraps(func)
    async def do(*args: TParams.args, **kwargs: TParams.kwargs) -> TResult:
        func_self: TOwner = kwargs.pop("self") if "self" in kwargs else owner  # type: ignore[assignment]
        if func_self is None:
            raise RuntimeError("Missing self argument.")

        return await func(func_self, *args, **kwargs)

    return do

sync_generator(func, owner) classmethod

Wrapper for synchronous generator methods.

Parameters:

Name Type Description Default
func Callable[Concatenate[TOwner, TParams], Generator[TResult, Any, Any]]

The function to wrap.

 required
owner TOwner | None

An optional owner object.

 required

Returns:

Type Description
Callable[TParams, Generator[TResult, None, None]]

The wrapper.
Source code in fasted/selfdependent.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@classmethod
def sync_generator(
    cls,
    func: Callable[Concatenate[TOwner, TParams], Generator[TResult, Any, Any]],
    owner: TOwner | None,
) -> Callable[TParams, Generator[TResult, None, None]]:
    """
    Wrapper for synchronous generator methods.

    Arguments:
        func: The function to wrap.
        owner: An optional owner object.

    Returns:
        The wrapper.
    """

    @wraps(func)
    def do(*args: TParams.args, **kwargs: TParams.kwargs) -> Generator[TResult, None, None]:
        func_self: TOwner = kwargs.pop("self") if "self" in kwargs else owner  # type: ignore[assignment]
        if func_self is None:
            raise RuntimeError("Missing self argument.")

        yield from func(func_self, *args, **kwargs)

    return do

sync_method(func, owner) classmethod

Wrapper for synchronous methods.

Parameters:

Name Type Description Default
func Callable[Concatenate[TOwner, TParams], TResult]

The function to wrap.

 required
owner TOwner | None

An optional owner object.

 required

Returns:

Type Description
Callable[TParams, TResult]

The wrapper.
Source code in fasted/selfdependent.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
@classmethod
def sync_method(
    cls,
    func: Callable[Concatenate[TOwner, TParams], TResult],
    owner: TOwner | None,
) -> Callable[TParams, TResult]:
    """
    Wrapper for synchronous methods.

    Arguments:
        func: The function to wrap.
        owner: An optional owner object.

    Returns:
        The wrapper.
    """

    @wraps(func)
    def do(*args: TParams.args, **kwargs: TParams.kwargs) -> TResult:
        func_self: TOwner = kwargs.pop("self") if "self" in kwargs else owner  # type: ignore[assignment]
        if func_self is None:
            raise RuntimeError("Missing self argument.")

        return func(func_self, *args, **kwargs)

    return do