Merge pull request #1 from Nikakto/0.2.0

0.2.0

README.md

@@ -0,0 +1,130 @@
+![CDNJS](https://img.shields.io/badge/Python-3.8%20%7C%203.9%20%7C%203.10%20%7C%203.11-2334D058)
+![CDNJS](https://shields.io/badge/FastAPI-%3E=0.7.0-009485)
+
+# FastAPI depends extension
+
+## Introduction
+
+Sometimes your FastAPI dependencies have to get value from functions cannot be available on initialization. The problem is particularly acute to use class dependencies with inheritance. This project try to solve problem of modify `Depends` after application initialization.
+
+## Installation
+
+```
+pip install fastapi-depends-ext
+```
+
+## Tutorial
+
+#### DependsAttr
+
+```python
+from typing import List
+
+from fastapi import Depends
+from fastapi import FastAPI
+from fastapi import Query
+from pydantic import conint
+
+from fastapi_depends_ext import DependsAttr
+from fastapi_depends_ext import DependsAttrBinder
+
+
+class ItemsPaginated(DependsAttrBinder):
+    _items = list(range(100))
+
+    async def get_page(self, page: conint(ge=1) = Query(1)):
+        return page
+
+    async def items(self, page: int = DependsAttr("get_page")):
+        _slice = slice(page * 10, (page + 1) * 10)
+        return self._items[_slice]
+
+
+class ItemsSquarePaginated(ItemsPaginated):
+    async def items(self, items: List[int] = DependsAttr("items", from_super=True)):
+        return [i**2 for i in items]
+
+
+app = FastAPI()
+
+
+@app.get("/")
+def items_list(items: List[int] = Depends(ItemsPaginated().items)) -> List[int]:
+    return items
+
+
+@app.get("/square")
+def items_list_square(items: List[int] = Depends(ItemsSquarePaginated().items)) -> List[int]:
+    return items
+```
+
+Use `DependsAttr` to `Depends` from current instance attributes. All examples use `asyncio`, but you can write all methods synchronous.
+
+`DependsAttr` support next properties:
+- class variables (must contains `callable` object)
+- class methods
+- static methods
+- instance methods
+- `property` returning `callable`
+
+Your class must inherit from `DependsAttrBinder` and attributes must be `DependsAttr`. `DependsAttrBinder` automatically patch all methods with `DependsAttr` by instance attributes.
+
+`DependsAttr` arguments:
+- `method_name` - `str`, name of instance attribute to use as dependency
+- `from_super` - `bool`, on true, will use attribute `method_name` from super class like `super().method_name()`
+- `use_cache` - `bool`, allow to cache depends result for the same dependencies in request
+
+#### DependsExt
+
+Useless(?) class created to proof of concept of patching methods and correct work `FastAPI` applications.
+
+`DependsExt` allow you define default values of method arguments after `FastAPI` endpoint has been created.  
+
+Example:
+```
+from fastapi import FastAPI
+from fastapi import Query
+
+from fastapi_depends_ext import DependsExt
+
+
+def pagination(page: int = Query()):
+    return page
+
+
+depends = DependsExt(pagination)
+
+
+app = FastAPI()
+
+
+@app.on_event("startup")
+def setup_depends():
+    depends.bind(page=Query(1))
+
+
+@app.get("/")
+def get_method(value: int = depends) -> int:
+    return value
+
+```
+
+Is equivalent for
+```
+from fastapi import Depends
+from fastapi import FastAPI
+from fastapi import Query
+
+
+def pagination(page: int = Query(1)):
+    return page
+
+
+app = FastAPI()
+
+
+@app.get("/")
+def get_method(value: int = Depends(pagination)) -> int:
+    return value
+
+```

fastapi_depends_ext/__init__.py

@@ -1,3 +1,3 @@
 from .depends import DependsExt
-from .depends import DependsMethod
-from .depends import DependsMethodBinder
+from .depends import DependsAttr
+from .depends import DependsAttrBinder

fastapi_depends_ext/depends.py

@@ -1,3 +1,4 @@
+import functools
 import inspect
 from typing import Any
 from typing import Callable
@@ -14,55 +15,63 @@
 
 
 SUPPORTED_DEPENDS = Union[Callable[..., Any], FieldInfo, params.Depends]
-SPECIAL_METHODS: Final = ("__call__", "__init__", "__new__")
+SPECIAL_METHODS_ERROR: Final = ("__call__",)
+SPECIAL_METHODS_IGNORE: Final = ("__init__", "__new__")
 
 
-class DependsMethodBinder:
+class DependsAttrBinder:
     def __init__(self, *args, **kwargs):
-        super(DependsMethodBinder, self).__init__(*args, **kwargs)
+        super(DependsAttrBinder, self).__init__(*args, **kwargs)
 
         methods = inspect.getmembers(self, predicate=inspect.ismethod)
         functions = inspect.getmembers(self, predicate=inspect.isfunction)
 
         for method_name, method in methods + functions:
+            if method_name in SPECIAL_METHODS_IGNORE:
+                continue
+
             signature = get_typed_signature(method)
-            values_is_depends_method = [
-                param.default for param in signature.parameters.values() if isinstance(param.default, DependsMethod)
+            values_is_depends_attr = [
+                param.default for param in signature.parameters.values() if isinstance(param.default, DependsAttr)
             ]
 
-            if not values_is_depends_method:
+            if not values_is_depends_attr:
                 continue
 
-            if method_name in SPECIAL_METHODS:
+            if method_name in SPECIAL_METHODS_ERROR:
                 class_method = f"{type(self).__name__}.{method.__name__}"
-                raise AttributeError(f"`{class_method}` can't have `DependsMethod` as default value for arguments")
+                raise AttributeError(f"`{class_method}` can't have `DependsAttr` as default value for arguments")
 
             self.bind(method)
 
     def bind(self, method: Callable) -> Callable:
-        def depends_method_bind(depends: DependsMethod, _base_class: type, instance) -> DependsMethod:
+        def depends_attr_bind(depends: DependsAttr, _base_class: type, instance) -> DependsAttr:
 
-            # todo: DependsMethod.__copy__
-            depends_copy = DependsMethod(
+            # todo: DependsAttr.__copy__
+            depends_copy = DependsAttr(
                 method_name=depends.method_name,
                 from_super=depends.from_super,
                 use_cache=depends.use_cache,
             )
 
-            dependency = depends_method_get_method(depends, _base_class, instance)
-            depends_copy.dependency = self.bind(dependency)
+            method_definition = getattr(_base_class, depends.method_name)
+            if isinstance(method_definition, property):
+                depends_copy.dependency = functools.partial(method_definition.fget, instance)
+            else:
+                dependency = depends_attr_get_method(depends, _base_class, instance)
+                depends_copy.dependency = self.bind(dependency)
             return depends_copy
 
-        def depends_method_get_method(depends: DependsMethod, _base_class: type, instance) -> Callable:
-            # todo: DependsMethod.get_method
+        def depends_attr_get_method(depends: DependsAttr, _base_class: type, instance) -> Callable:
+            # todo: DependsAttr.get_method
             obj = super(_base_class, instance) if depends.from_super else instance
             return getattr(obj, depends.method_name)
 
-        base_class = get_base_class(method)
+        base_class = get_base_class(self, method.__name__, method)
         signature = get_typed_signature(method)
-        parameters = (param for param in signature.parameters.values() if isinstance(param.default, DependsMethod))
+        parameters = (param for param in signature.parameters.values() if isinstance(param.default, DependsAttr))
         instance_method_params = {
-            parameter.name: depends_method_bind(parameter.default, base_class, self) for parameter in parameters
+            parameter.name: depends_attr_bind(parameter.default, base_class, self) for parameter in parameters
         }
 
         if instance_method_params:
@@ -91,9 +100,9 @@ def bind(self, **kwargs: SUPPORTED_DEPENDS) -> "DependsExt":
         return DependsExt(patched, use_cache=self.use_cache)
 
 
-class DependsMethod(DependsExt):
+class DependsAttr(DependsExt):
     def __init__(self, method_name: str, *, from_super: bool = False, use_cache=True):
-        super(DependsMethod, self).__init__(use_cache=use_cache)
+        super(DependsAttr, self).__init__(use_cache=use_cache)
         self.from_super = from_super
         self.method_name = method_name
 
@@ -119,12 +128,12 @@ def bind(self, instance, super_from: type = None):
                 cls_name = f"super({cls_name}, instance)"
             raise AttributeError(f"{cls_name} has not method `{self.method_name}`")
 
-        cls = get_base_class(method)
+        cls = get_base_class(instance, self.method_name, method)
         signature = get_typed_signature(method)
 
         for parameter in signature.parameters.values():
-            depends: DependsMethod = parameter.default
-            if not isinstance(depends, DependsMethod) or depends.is_bound:
+            depends: DependsAttr = parameter.default
+            if not isinstance(depends, DependsAttr) or depends.is_bound:
                 continue
 
             elif depends.method_name != self.method_name or depends.from_super:

fastapi_depends_ext/utils.py

@@ -4,22 +4,32 @@
 from types import FunctionType
 from types import MethodType
 from typing import Callable
+from typing import Optional
 
 from fastapi.dependencies.utils import get_typed_signature
 
 
-def get_base_class(method: Callable) -> type:
-    if inspect.ismethod(method):
-        for cls in inspect.getmro(method.__self__.__class__):
-            if cls.__dict__.get(method.__name__) is method.__func__:
-                return cls
-        method = method.__func__  # fallback to __qualname__ parsing
+def _get_func(instance, func) -> callable:
+    if type(func) is property:
+        return _get_func(instance, func.fget(instance))
+    elif type(func) in (classmethod, staticmethod) or inspect.ismethod(func):
+        return func.__func__
+    elif callable(func):
+        return func
+    else:
+        raise TypeError(f"Incorrect type of `{func}`")
 
-    if inspect.isfunction(method):
-        cls = getattr(inspect.getmodule(method), method.__qualname__.split(".<locals>", 1)[0].rsplit(".", 1)[0])
-        if isinstance(cls, type):
+
+def get_base_class(instance: object, method_name: str, method_target: [type, Callable]) -> Optional[type]:
+    for cls in inspect.getmro(type(instance)):
+        method_cls = cls.__dict__.get(method_name)
+        if method_cls is None:  # check to None cause can be not callable like property object (not property value)
+            continue
+
+        _func_class = _get_func(instance, method_cls)
+        _func_target = _get_func(instance, method_target)
+        if _func_class is _func_target:
             return cls
-    return getattr(method, "__objclass__", None)  # handle special descriptor objects
 
 
 def get_super_for_method(base_class: type, method_name: str, super_from: type = None) -> type: