Недавно решил переписать часть проекта, связанную с запросами к API маркетплейсов, используя асинхронщину. При этом решил сохранить старый, синхронный вариант. В голову пришла идея на выходе получить класс, в который будет передаваться, скажем, тип/мод SyncApi или AsyncApi и в зависимости от переданного параметра запросы будут синхронными/асинхронными соответственно.
Вот пример, что имеется и чего хотелось бы достичь:
# Как это выглядело изначально:
class Api:
def __init__(self, base_url: AnyStr = None, headers: Dict = None) -> None:
self.__base_url: str = base_url
self.__headers: Dict = headers
@property
def base_url(self):
...
@property
def headers(self):
...
def request(self):
...
class OzonApi(Api):
__base_url: str = 'https://api-seller.ozon.ru'
__headers: Dict = {
'Host': 'api-seller.ozon.ru',
'Client-Id': CLIENT_ID,
'Api-Key': API_KEY,
'Content-Type': 'application/json'
}
def __init__(self) -> None:
super().__init__(base_url=self.__base_url, headers=self.__headers)
def get_category_tree(self):
_method: str = '/v1/description-category/tree'
_json_body: Dict = {}
resp: List[Dict] = self.request()
return resp
def get_orders(self):
...
def get_actions(self):
...
# Такая же реализация
class WbStandartApi(Api):
...
На выходе получались классы с синхронными запросами к API, данные которых шли далее по пайплайну.
# Сейчас это выглядит так:
class BaseApi:
@property
@abstractmethod
def domain(self):
return NotImplementedError()
@abstractmethod
def fetch(self):
return NotImplementedError()
@abstractmethod
def fetch_with_offset(self):
return NotImplementedError()
@abstractmethod
def fetch_with_paging(self):
return NotImplementedError()
@abstractmethod
def fetch_keep_last(self):
return NotImplementedError()
class AsyncApi(BaseApi):
def __init__(
self,
domain: AnyStr = None,
headers: Dict = None,
method: str = None,
endpoint: str = None,
dataclass: Callable = None
) -> None:
self.__domain: str = domain
self.__headers: Dict = headers
self.__method: str = method
self.__endpoint: str = endpoint
self.__dataclass: Callable = dataclass
self._session: ClientSession = None
self.data: List[Dict] = []
async def fetch(self) -> str:
return 'This is fetch with async api'
async def fetch_with_offset(self) -> str:
return 'This is fetch_with_offset with async api'
async def fetch_with_paging(self) -> str:
return 'This is fetch_with_paging with async api'
async def fetch_keep_last(self) -> str:
return 'This is fetch_keep_last with async api'
class SyncApi(BaseApi):
def __init__(
self,
domain: AnyStr = None,
headers: Dict = None,
method: str = None,
endpoint: str = None
) -> None:
self.__domain: str = domain
self.__headers: Dict = headers
self.__method: str = method
self.__endpoint: str = endpoint
self.__dataclass: Callable = None
self._session: Session = None
self.data: List[Dict] = []
@property
def domain(self) -> bytes:
return b'This is domain sync api'
def fetch(self) -> bytes:
return b'This is fetch with sync api'
def fetch_with_offset(self) -> bytes:
return b'This is fetch_with_offset with sync api'
def fetch_with_paging(self) -> bytes:
return b'This is fetch_with_paging with sync api'
def fetch_keep_last(self) -> bytes:
return b'This is fetch_keep_last with sync api'
ApiT = Union[SyncApii, AsyncApii]
class OzonSellerApi:
_domain = ''
_headers = ''
def __init__(self, engine: SyncApii | AsyncApii):
self.engine: SyncApii | AsyncApii = engine(
domain=self._domain,
headers=self._headers,
method=self.method,
endpoint=self.endpoint,
dataclass=self.dataclass
)
@property
def endpoint(self):
...
@property
def method(self):
...
@property
def dataclass(self):
...
class OzonActionApi(OzonSellerApi):
def __init__(self, engine: ApiT):
super().__init__(engine)
def get_actions(self):
self.endpoint = '/endpoint'
self.method = 'POST'
self.json = {}
self.dataclass = Actions
if self.engine == SyncApii:
return self.engine.fetch_with_offset() # Sync request
if self.engine == AsyncApii:
return self.engine.fetch_with_offset() # Async request
class OzonOrdersApi(OzonSellerApi):
def __init__(self, engine: ApiT):
super().__init__(engine)
def get_orders(self):
self.endpoint = '/endpoint2'
self.method = 'POST'
self.json = {}
self.dataclass = Orders
if self.engine == SyncApii:
return self.engine.fetch_with_offset() # Sync request
if self.engine == AsyncApii:
return self.engine.fetch_with_offset() # Async request
class OzonSellerApiAdapter:
def __init__(self, engine: ApiT):
self.actions = OzonActionApi(engine)
self.orders = OzonOrdersApi(engine)
class OzonFabric:
def __init__(self, engine: ApiT):
self.seller = OzonSellerApiAdapter(engine)
sf = OzonFabric(SyncApii).seller.actions.get_actions() # Sync request
af = OzonFabric(AsyncApii).seller.actions.get_actions() # Async request
Я прекрасно понимаю, что совмещение в одном классе синхронных и асинхронных функций это не лучший способ реализации, однако когда я разделял наследование синхронных и асинхронных, то код во многом дублировался, что я посчитал плохим знаком.
Суть вопроса заключается в том, как лучше построить дерево наследования при использовании синхронного базового класса и асинхронного? По сути отличается во всем этом дереве это базовые классы и способы запроса к серверу, все остальные одинаковое. Возможно кто-то сталкивался с таким вопросом или читал что-либо, что может помочь?
