🧑‍🏫
liualexiang
  • Introduction
  • Azure
    • AKS Basic
    • AKS Spark
    • AZ ACR SYNC
    • Azure CMI SDWAN
    • Azure LB DeepDive
      • Azure LB DeepDive
    • Azure Service Principal Basic
    • Azure Internal VM Network Connectivity
      • Azure Internal VM Network Connectivity
    • Azure Cli Build
    • Azure Vm Memory Monitor
  • Blockchain
    • BTC
  • CRISPR
    • 使用Parallel_Cluster提升CRISPA效率
  • OpenSource
    • ElasticSearch
      • ES Get Started
      • ES Search Query
      • Kibana 可视化
      • Logstash配置
    • Ansible 基础
    • Infra As Code
      • Pulumi Get Started
      • Terraform Basic
    • ZooKeeper 基础
    • RPC与REST
    • 使用Python申请大量内存测试
    • 使用TPC_DS产生压测数据
    • Superset
      • Superset部署手册
    • 代码扫描
    • Git
      • Git Basic
      • Github Action Basic
      • Gitlab与AzureAD集成
      • Gitbook 基础教程
    • K8S
      • enter_node
      • K8s X509 Client Cert
      • K8s Basic
      • K8s Oidc
      • Docker 基础
      • helm基础
      • K8S_Secrets管理
      • 深入了解K8S
      • 混沌工程
      • Istio
      • 生态
      • CRD开发
      • k8s网络
    • Cloud_Custodian
    • Jenkins Basic
    • Nginx
    • ETCD
    • 正则
    • VictoriaMetrics
    • Kafka
  • MySQL
    • MySQL 调优
  • Linux
    • SSH Tunnel 上网
    • 内存管理
    • 在Linux系统中通过LUKS加密磁盘
    • 量子计算 Basic
    • IO多路复用
    • Iptables
    • tmux和screen
    • Systemd
    • OS 基础
    • jq基础
    • yum
    • neovim
  • Web
    • Html Css
    • Web部署
    • 缓存
  • Programming
    • 算法
      • 返回list中最大生序子序列长度
    • Python技巧
      • Python的语法糖
      • Python常用装饰器
      • AsyncIO基础
      • 自动化测试pytest
      • python中的下划线
      • 面向对象
      • Python的坑
      • Python配置文件管理
      • HTTP Stream Response
      • Python项目管理
    • 设计模式
      • 设计模式
      • 面向对象的思想
      • 编程概念
    • Go
      • Go 基础
      • Go常用功能
      • 结构体入门
    • 前端
    • Vue
    • NodeJS
  • Math
    • 多项式插值法
  • Security
    • HTTP常见攻击
    • 加密与签名
    • RSA
    • ECDSA
  • Solidity
    • Solidity基础
    • Blockchain Testnet Faucet
  • Tools
    • 视频处理ffmpeg
    • IDE配置
    • iTerm2美化
    • 密码管理
    • FRP配置
    • 工具集
由 GitBook 提供支持
在本页
  1. Programming
  2. Python技巧

面向对象

上一页python中的下划线下一页Python的坑

最后更新于1年前

使用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)
inject