面向对象

使用inject库来实现依赖注入

inject 库是一个第三方库，在初始化 DB或者 redis等连接的时候，往往我们希望在一个地方初始化了之后，其他地方引用的时候，能够直接引用，而不需要再次对数据库或者缓存文件进行连接。而且有可能在多个python文件里，都会对数据库或者缓存进行操作。那么 inject 库就可以很好的实现单例模式。

用 ElasticSearch 的连接作为一个示例。比如有 es_conn.py文件，代码如下:

import boto3
from aws_requests_auth.aws_auth import AWSRequestsAuth
from elasticsearch import Elasticsearch, RequestsHttpConnection
import inject


def init_es(region: str, domain: str) -> Elasticsearch:
    _credentials = boto3.Session().get_credentials()
    _awsauth = AWSRequestsAuth(
        aws_region=region,
        aws_service="es",
        aws_access_key=_credentials.access_key,
        aws_secret_access_key=_credentials.secret_key,
        aws_token=_credentials.token,
        aws_host=domain
    )
    return Elasticsearch(
        host=domain,
        http_auth=_awsauth,
        port=443,
        use_ssl=True,
        connection_class=RequestsHttpConnection,
        verify_certs=False
    )


class LogES(Elasticsearch):
    """
    Why using the empty class?
    if there was multiple ES connection, each connection will use one empty class,
    by using this way to confirm the ES connection can be only init once.
    """
    pass


@inject.configure
def config(binder: inject.Binder):
    es_init = init_es(domain="abc.es.amazonaws.com", region="ap-northeast-1")
    binder.bind(LogES, es_init)

此时在其他文件里，可以直接通过 inject.instance 来调用之前实例化之后的对象

import inject
from es_conn import LogES

if __name__ == "__main__":
    ## invoke it anywhere
    es = inject.instance(LogES)
    result = es.search(index="*access-log*", body={})
    print(result)

值得关注的是：在上述 es_conn 的代码中，我们看到 class LogES 这个class并未实现任何方法，如果只有一个ES的连接，其实这个class是没必要存在的，我们在 binder.bind的时候，直接将 es_init 绑定到 ElasticSearch 这个类上就可以了。但是如果我们有多个ES连接，那么需要对每一个ES连接都创建单独的一个类，这样binder.bind 的时候，可以将每一个实例化的对象都bind到一个类上，以此实现多个连接的单例.

扩展:

在上述代码中，我们发现 init_es 函数是直接在外面定义的，如果想要放在类内，但是又想直接让类初始化的时候，就能获得实例化对象的 search，query等方法，那么我们可以将初始化连接的部分代码，放到init函数里，但是由于init函数没有返回值，为了让我们的这个类有 ElasticSearch 的所有方法，可以用 __getattr__这个方法来实现。当执行类里某个不存在的方法的时候，就会执行 __getattr__ 这个方法，而这个方法执行的逻辑是用 getattr() 方法查找 self._client 里的这个方法，因此就实现了我们自定义 ES 类有 ElasticSearch类的所有方法的能力

class ES:
    def __init__(self, region: str, domain: str) -> Elasticsearch:
        _credentials = boto3.Session().get_credentials()
        _awsauth = AWSRequestsAuth(
            aws_region=region,
            aws_service="es",
            aws_access_key=_credentials.access_key,
            aws_secret_access_key=_credentials.secret_key,
            aws_token=_credentials.token,
            aws_host=domain
        )

        self._client = Elasticsearch(
            host=domain,
            http_auth=_awsauth,
            port=443,
            use_ssl=True,
            connection_class=RequestsHttpConnection,
            verify_certs=False
        )
        
    def __getattr__(self, item):
        try:
            return getattr(self._client, item)
        except AttributeError:
            raise AttributeError(f"{self.__class__.__name__} object has no attribute {item}")
            
# 也可以用 inject 做依赖注入，这里为了简化就不写了
client = ES(region="ap-northeast-1", domain="abc.es.amazonaws.com")
response = client.search(index="*cserver*",body={})
print(response)