mirror of https://gitee.com/anolis/sysom.git
131 lines
4.0 KiB
Python
131 lines
4.0 KiB
Python
# -*- coding: utf-8 -*- #
|
|
"""
|
|
Time 2023/5/8 19:51
|
|
Author: mingfeng (SunnyQjm)
|
|
Email mfeng@linux.alibaba.com
|
|
File gclient.py
|
|
Description:
|
|
"""
|
|
import requests
|
|
import importlib
|
|
from threading import Lock
|
|
from clogger import logger
|
|
from abc import ABCMeta, abstractmethod
|
|
from .exceptions import GClientProtoAlreadyExistsException, \
|
|
GClientProtoNotExistsException
|
|
from .url import GClientUrl
|
|
|
|
|
|
class GClient(metaclass=ABCMeta):
|
|
proto_dict = {}
|
|
proto_lock = Lock()
|
|
|
|
@abstractmethod
|
|
def get(self, path: str, params=None,
|
|
**kwargs) -> requests.Response:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def options(self, path: str, **kwargs) -> requests.Response:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def head(self, path: str, **kwargs) -> requests.Response:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def post(self, path: str, data=None, json=None,
|
|
**kwargs) -> requests.Response:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def put(self, path: str, data=None,
|
|
**kwargs) -> requests.Response:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def patch(self, path: str, data=None,
|
|
**kwargs) -> requests.Response:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def delete(self, path: str, **kwargs) -> requests.Response:
|
|
pass
|
|
|
|
@staticmethod
|
|
def protocol_register(proto, sub_class):
|
|
"""Register one new protocol => indicate one execution module
|
|
|
|
Register a new protocol => This function is called by the executing
|
|
module to register its own implementation of Producer for the executing
|
|
module to take effect.
|
|
(Usually when the execution module is implemented according to the
|
|
specification, there is no need for the developer to call this method
|
|
manually, the abstraction layer will dynamically import)
|
|
|
|
Args:
|
|
proto(str): Protocol identification
|
|
sub_class: Implementation class of Producer
|
|
|
|
Returns:
|
|
|
|
Examples:
|
|
>>> GClient.protocol_register('redis', HttpGClient)
|
|
|
|
"""
|
|
if proto in GClient.proto_dict:
|
|
err = GClientProtoAlreadyExistsException(
|
|
f"Proto '{proto}' already exists in Cmg-base-GClient."
|
|
)
|
|
logger.error(err)
|
|
raise err
|
|
GClient.proto_dict[proto] = sub_class
|
|
logger.info(
|
|
f"GClient-base-GClient register proto '{proto}' success"
|
|
)
|
|
|
|
|
|
def dispatch_g_client(url: str, **kwargs) -> GClient:
|
|
"""Construct one GClient instance according the url
|
|
|
|
Construct a GClient instance of the corresponding type based on
|
|
the URL passed in.
|
|
|
|
Args:
|
|
url(str): GClientUrl
|
|
|
|
Returns:
|
|
GClient: one GClient instance
|
|
|
|
Examples:
|
|
>>> g_client = dispatch_g_client(
|
|
"cache_name",
|
|
..."http://localhost:6379")
|
|
"""
|
|
g_client_url = GClientUrl.parse(url)
|
|
with GClient.proto_lock:
|
|
if g_client_url.proto not in GClient.proto_dict:
|
|
# Check if dynamic import is possible
|
|
target_module = f"gclient_{g_client_url.proto}.{g_client_url.proto}_gclient"
|
|
try:
|
|
module = importlib.import_module(target_module)
|
|
GClient.protocol_register(
|
|
g_client_url.proto,
|
|
getattr(module, f'{g_client_url.proto.capitalize()}GClient')
|
|
)
|
|
except ModuleNotFoundError as exc:
|
|
logger.error(
|
|
f"Try to auto import module {target_module} failed."
|
|
)
|
|
raise GClientProtoNotExistsException(
|
|
f"Proto '{g_client_url.proto}' not exists in GClient-base"
|
|
"-GClient."
|
|
) from exc
|
|
g_client_instance = GClient.proto_dict[g_client_url.proto](g_client_url,
|
|
**kwargs)
|
|
logger.info(
|
|
f"GClient-base-GClient dispatch one GClient instance "
|
|
f"success. proto={g_client_url.proto}, url={url}"
|
|
)
|
|
return g_client_instance
|