Skip to content

Instantly share code, notes, and snippets.

@Tishka17
Last active December 21, 2022 16:41
Show Gist options
  • Save Tishka17/b408565cfdf96503f899559d89de4721 to your computer and use it in GitHub Desktop.
Save Tishka17/b408565cfdf96503f899559d89de4721 to your computer and use it in GitHub Desktop.
Simple IoC framework
import contextlib
from contextlib import AbstractContextManager, ExitStack
from functools import wraps
from inspect import getfullargspec
from typing import Any, Callable, Dict, get_type_hints, Type
class Depends: # todo mypy
def __init__(self, dependency: Any = None):
self.dependency = dependency
Dependency = Callable[[], AbstractContextManager]
def static_dependency(value):
@contextlib.contextmanager
def manager():
yield value
return manager
STUB = object()
def lazy_static_dependency(factory, *args, **kwargs):
value = STUB
@contextlib.contextmanager
def manager():
nonlocal value
if value is STUB:
value = factory(*args, **kwargs)
yield value
return manager
class InjectionMiddleware:
def __init__(self, overrides: Dict[Callable, Dependency]):
self.overrides = overrides
def _normalize_dependecy(self, type_: Type, depends: Depends):
if depends.dependency is None:
return type_
return depends.dependency
def _get_dependencies(
self, func,
) -> Dict[str, Callable]:
hints = get_type_hints(func)
argspec = getfullargspec(func)
defaults = argspec.defaults or []
kwonlydefaults = argspec.kwonlydefaults or {}
positional = {
arg: self._normalize_dependecy(hints.get(arg), default)
for arg, default in zip(argspec.args[::-1], defaults[::-1])
if isinstance(default, Depends)
}
keyword = {
arg: self._normalize_dependecy(hints.get(arg), default)
for arg, default in kwonlydefaults.items()
if isinstance(default, Depends)
}
return {
**positional,
**keyword,
}
def handler(self, handler):
required_dependencies = self._get_dependencies(handler)
@wraps(handler)
def wrapper(*args, **kwargs):
stack = ExitStack()
dependencies = {}
for name, dependency in required_dependencies.items():
dependency_context = self.overrides[dependency]()
dependencies[name] = stack.enter_context(dependency_context)
try:
kwargs.update(dependencies)
return handler(*args, **kwargs)
finally:
stack.close()
return wrapper
from fastapi import FastAPI, Response, Request
from depends import Depends, static_dependency, InjectionMiddleware
def hello_world(request: Request, answer: str = Depends()):
print(answer)
return Response(answer)
app = FastAPI()
m = InjectionMiddleware(overrides={
str: static_dependency("hello")
})
app.add_route("/", route=m.handler(hello_world))
from flask import Flask
from depends import Depends, static_dependency, InjectionMiddleware
def hello_world(answer: str = Depends()):
return answer
app = Flask(__name__)
m = InjectionMiddleware(overrides={
str: static_dependency("hello")
})
app.add_url_rule("/", view_func=m.handler(hello_world))
app.run()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment