依赖注入实现中

fund_pool/fund_optimize.py

@@ -14,6 +14,7 @@ def find_optimize(fund_ids, day):
         raise NameError(f"find optimize, but not found sortino config.")
     start = filter_weekend(sorted([day - relativedelta(days=1, **dict_remove(x, ('weight', 'name'))) for x in sortino_config])[0])
     fund_navs = pd.DataFrame(navs.get_navs(fund_id=tuple(fund_ids), min_date=start, max_date=day))
+
     fund_navs.sort_values('nav_date', inplace=True)
     fund_navs = fund_navs.pivot_table(index='nav_date', columns='fund_id', values='nav_cal')
     fund_navs.fillna(method='ffill', inplace=True)

utils/__init__.py

@@ -3,3 +3,6 @@ from .base import *
 from .datebase import read, write, transaction, where
 from .__env_config import config, get_config
 from .__logger import build_logger, logger
+from .injectable import component
+
+del injectable, __logger, __env_config, datebase, base, date_utils

utils/injectable.py

+import abc
+import inspect
+import functools
+from typing import List, get_origin, get_args
+from types import GenericAlias
+
+__COMPONENT_CLASS = []
+__NAME_COMPONENT = {}
+__COMPONENT_INSTANCE = {}
+
+
+class InjectableError(Exception):
+    def __init__(self, msg):
+        self.__msg = msg
+
+    def __str__(self):
+        return self.__msg
+
+
+def component(cls=None, bean_name=None):
+    if cls is None:
+        return functools.partial(component, bean_name=bean_name)
+
+    __COMPONENT_CLASS.append(cls)
+    if bean_name:
+        if bean_name in __NAME_COMPONENT:
+            raise InjectableError(f"bean name[{bean_name}] is already defined.")
+        __NAME_COMPONENT[bean_name] = cls
+    return cls
+
+
+def autowired(func=None):
+    if func is None:
+        return functools.partial(autowired)
+
+    @functools.wraps(func)
+    def wrap(*args, **kwargs):
+        self_type = type(args[0])
+        if self_type in __COMPONENT_CLASS and self_type not in __COMPONENT_INSTANCE:
+            __COMPONENT_INSTANCE[self_type] = args[0]
+        for p_name, p_type in inspect.signature(func).parameters.items():
+            if p_name == 'self' or p_type == inspect.Parameter.empty:
+                continue
+            if get_origin(p_type.annotation) is list:
+                inject_types = get_args(p_type.annotation)
+                if len(inject_types) > 0:
+                    instances = [get_instance(x) for x in __COMPONENT_CLASS if issubclass(x, inject_types)]
+                    if len(instances) > 0:
+                        kwargs[p_name] = instances
+            else:
+                components = [x for x in __COMPONENT_CLASS if issubclass(x, p_type.annotation)]
+                if len(components) > 0:
+                    kwargs[p_name] = get_instance(components[0])
+        func(*args, **kwargs)
+
+    return wrap
+
+
+def get_instance(t):
+    if t not in __COMPONENT_CLASS:
+        return None
+    if t not in __COMPONENT_INSTANCE:
+        __COMPONENT_INSTANCE[t] = t()
+    return __COMPONENT_INSTANCE[t]
+
+
+class BaseBean(metaclass=abc.ABCMeta):
+    @abc.abstractmethod
+    def do_something(self):
+        pass
+
+
+@component(bean_name="one")
+class OneImplBean(BaseBean):
+
+    def do_something(self):
+        print("one bean")
+
+
+@component(bean_name="two")
+class TwoImplBean(BaseBean):
+
+    def do_something(self):
+        print("two bean")
+
+
+class ServiceBean:
+    @autowired
+    def __init__(self, base: BaseBean = None, bases: List[BaseBean] = None):
+        base.do_something()
+        for b in bases:
+            b.do_something()
+
+
+
+
+print(__COMPONENT_CLASS, __NAME_COMPONENT)
+
+if __name__ == '__main__':
+    ServiceBean()