Spaces:
Runtime error
Runtime error
File size: 2,831 Bytes
58d33f0 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 |
"""Lightweight wrapper around requests library, with async support."""
from typing import Any, Dict, Optional
import aiohttp
import requests
from pydantic import BaseModel, Extra
class RequestsWrapper(BaseModel):
"""Lightweight wrapper around requests library."""
headers: Optional[Dict[str, str]] = None
aiosession: Optional[aiohttp.ClientSession] = None
class Config:
"""Configuration for this pydantic object."""
extra = Extra.forbid
arbitrary_types_allowed = True
def get(self, url: str) -> str:
"""GET the URL and return the text."""
return requests.get(url, headers=self.headers).text
def post(self, url: str, data: Dict[str, Any]) -> str:
"""POST to the URL and return the text."""
return requests.post(url, json=data, headers=self.headers).text
def patch(self, url: str, data: Dict[str, Any]) -> str:
"""PATCH the URL and return the text."""
return requests.patch(url, json=data, headers=self.headers).text
def put(self, url: str, data: Dict[str, Any]) -> str:
"""PUT the URL and return the text."""
return requests.put(url, json=data, headers=self.headers).text
def delete(self, url: str) -> str:
"""DELETE the URL and return the text."""
return requests.delete(url, headers=self.headers).text
async def _arequest(self, method: str, url: str, **kwargs: Any) -> str:
"""Make an async request."""
if not self.aiosession:
async with aiohttp.ClientSession() as session:
async with session.request(
method, url, headers=self.headers, **kwargs
) as response:
return await response.text()
else:
async with self.aiosession.request(
method, url, headers=self.headers, **kwargs
) as response:
return await response.text()
async def aget(self, url: str) -> str:
"""GET the URL and return the text asynchronously."""
return await self._arequest("GET", url)
async def apost(self, url: str, data: Dict[str, Any]) -> str:
"""POST to the URL and return the text asynchronously."""
return await self._arequest("POST", url, json=data)
async def apatch(self, url: str, data: Dict[str, Any]) -> str:
"""PATCH the URL and return the text asynchronously."""
return await self._arequest("PATCH", url, json=data)
async def aput(self, url: str, data: Dict[str, Any]) -> str:
"""PUT the URL and return the text asynchronously."""
return await self._arequest("PUT", url, json=data)
async def adelete(self, url: str) -> str:
"""DELETE the URL and return the text asynchronously."""
return await self._arequest("DELETE", url)
|