Advertisement
Guest User

Untitled

a guest
May 31st, 2023
53
0
Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
Python 6.26 KB | Source Code | 0 0
  1. class BaseClient:
  2.     def __init__(self, token: str, org_id: str):
  3.         self._url = "https://api.tracker.yandex.net"
  4.         self._version = "v2"
  5.         self._params = RequestParams()
  6.         self._headers = {
  7.             "Authorization": f"OAuth {token}",lass BaseClient:
  8.     def __init__(self, token: str, org_id: str):
  9.         self._url = "https://api.tracker.yandex.net"
  10.         self._version = "v2"
  11.         self._params = RequestParams()
  12.         self._headers = {
  13.             "Authorization": f"OAuth {token}",
  14.             "X-Org-ID": org_id,
  15.         }
  16.         self._session = ClientSession(
  17.             base_url=f"{self._url}", raise_for_status=True, headers=self._headers
  18.         )
  19.  
  20.     async def __aenter__(self) -> "BaseClient":
  21.         return self
  22.  
  23.     async def __aexit__(
  24.         self,
  25.         exc_type: typing.Optional[typing.Type[BaseException]],
  26.         exc_val: typing.Optional[BaseException],
  27.         exc_tb: typing.Optional[TracebackType],
  28.     ) -> None:
  29.         await self._session.close()
  30.  
  31.     async def close(self):
  32.         await self._session.close()
  33.  
  34.     async def get(
  35.         self,
  36.         url: str,
  37.         params: typing.Dict[str, typing.Any] | None = None,
  38.     ):
  39.         default_params = self._params.dict(by_alias=True)
  40.         if params:
  41.             default_params.update(params)
  42.         return await self._session.get(f"/{self._version}/{url}", params=default_params)
  43.  
  44.     async def post(
  45.         self,
  46.         url: str,
  47.         data: TrackerModel | None,
  48.         params: typing.Dict[str, typing.Any] | None = None,
  49.     ) -> ClientResponse:
  50.         default_params = self._params.dict(by_alias=True)
  51.         if params:
  52.             default_params.update(params)
  53.         if data:
  54.             return await self._session.post(
  55.                 f"/{self._version}/{url}",
  56.                 params=params,
  57.                 data=data.json(by_alias=True, exclude_unset=True, exclude_none=True),
  58.             )
  59.         return await self._session.post(
  60.             f"/{self._version}/{url}", params=default_params
  61.         )
  62.  
  63.     async def patch(
  64.         self,
  65.         url: str,
  66.         data: TrackerModel,
  67.         params: typing.Dict[str, typing.Any] | None = None,
  68.     ) -> ClientResponse:
  69.         default_params = self._params.dict(by_alias=True)
  70.         if params:
  71.             default_params.update(params)
  72.         return await self._session.patch(
  73.             f"/{self._version}/{url}",
  74.             params=default_params,
  75.             data=data.json(by_alias=True, exclude_none=True),
  76.         )
  77.  
  78.     async def delete(
  79.         self,
  80.         url: str,
  81.         params: typing.Dict[str, typing.Any] | None = None,
  82.     ) -> ClientResponse:
  83.         return await self._session.delete(
  84.             f"/{self._version}/{url}",params=params
  85.         )
  86.  
  87.     async def put(
  88.         self,
  89.         url:str,
  90.         params: typing.Dict[str, typing.Any] | None = None,
  91.         data:TrackerModel| None = None
  92.     )->ClientResponse:
  93.         default_params = self._params.dict(by_alias=True)
  94.         if params:
  95.             default_params.update(params)
  96.         return await self._session.put(
  97.             f"/{self._version}/{url}",
  98.             params=default_params
  99.         )
  100.            
  101.        
  102.  
  103.     async def handle_response(
  104.         self,
  105.         response: ClientResponse,
  106.     ) -> ResponseParams:
  107.         return ResponseParams.parse_obj(response.headers)
  108.  
  109.             "X-Org-ID": org_id,
  110.         }
  111.         self._session = ClientSession(
  112.             base_url=f"{self._url}", raise_for_status=True, headers=self._headers
  113.         )
  114.  
  115.     async def __aenter__(self) -> "BaseClient":
  116.         return self
  117.  
  118.     async def __aexit__(
  119.         self,
  120.         exc_type: typing.Optional[typing.Type[BaseException]],
  121.         exc_val: typing.Optional[BaseException],
  122.         exc_tb: typing.Optional[TracebackType],
  123.     ) -> None:
  124.         await self._session.close()
  125.  
  126.     async def close(self):
  127.         await self._session.close()
  128.  
  129.     async def get(
  130.         self,
  131.         url: str,
  132.         params: typing.Dict[str, typing.Any] | None = None,
  133.     ):
  134.         default_params = self._params.dict(by_alias=True)
  135.         if params:
  136.             default_params.update(params)
  137.         return await self._session.get(f"/{self._version}/{url}", params=default_params)
  138.  
  139.     async def post(
  140.         self,
  141.         url: str,
  142.         data: TrackerModel | None,
  143.         params: typing.Dict[str, typing.Any] | None = None,
  144.     ) -> ClientResponse:
  145.         default_params = self._params.dict(by_alias=True)
  146.         if params:
  147.             default_params.update(params)
  148.         if data:
  149.             return await self._session.post(
  150.                 f"/{self._version}/{url}",
  151.                 params=params,
  152.                 data=data.json(by_alias=True, exclude_unset=True, exclude_none=True),
  153.             )
  154.         return await self._session.post(
  155.             f"/{self._version}/{url}", params=default_params
  156.         )
  157.  
  158.     async def patch(
  159.         self,
  160.         url: str,
  161.         data: TrackerModel,
  162.         params: typing.Dict[str, typing.Any] | None = None,
  163.     ) -> ClientResponse:
  164.         default_params = self._params.dict(by_alias=True)
  165.         if params:
  166.             default_params.update(params)
  167.         return await self._session.patch(
  168.             f"/{self._version}/{url}",
  169.             params=default_params,
  170.             data=data.json(by_alias=True, exclude_none=True),
  171.         )
  172.  
  173.     async def delete(
  174.         self,
  175.         url: str,
  176.         params: typing.Dict[str, typing.Any] | None = None,
  177.     ) -> ClientResponse:
  178.         return await self._session.delete(
  179.             f"/{self._version}/{url}",params=params
  180.         )
  181.  
  182.     async def put(
  183.         self,
  184.         url:str,
  185.         params: typing.Dict[str, typing.Any] | None = None,
  186.         data:TrackerModel| None = None
  187.     )->ClientResponse:
  188.         default_params = self._params.dict(by_alias=True)
  189.         if params:
  190.             default_params.update(params)
  191.         return await self._session.put(
  192.             f"/{self._version}/{url}",
  193.             params=default_params
  194.         )
  195.            
  196.        
  197.  
  198.     async def handle_response(
  199.         self,
  200.         response: ClientResponse,
  201.     ) -> ResponseParams:
  202.         return ResponseParams.parse_obj(response.headers)
  203.  
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement