Skip to content
Snippets Groups Projects
Commit 0bcc98f9 authored by Bjarke Madsen's avatar Bjarke Madsen
Browse files

use requests.Session to reuse TCP connections for Grafana API requests

parent 8b3e9bc3
No related branches found
No related tags found
No related merge requests found
import requests
import logging
from requests.models import HTTPError
from requests.adapters import HTTPAdapter
logger = logging.getLogger(__name__)
class Request(object):
class Request(requests.Session):
def __init__(self, url, headers=None):
self.headers = {
super().__init__()
# allow using up to 16 connections
adapter = HTTPAdapter(pool_maxsize=16)
self.mount(url, adapter)
self.headers.update({
'Accept': 'application/json'
}
})
if headers:
self.headers.update(headers)
self.BASE_URL = url
def do_request(self, method, endpoint, *args, **kwargs):
r = requests.request(method, self.BASE_URL + endpoint,
*args,
**kwargs,
headers={
**kwargs.get('headers', {}),
**self.headers
})
try:
r.raise_for_status()
except HTTPError as e:
if e.response.status_code < 500:
logger.error(e.response.content.decode('utf-8'))
raise e
return r.json()
def get(self, endpoint: str, *args, **kwargs):
return self.do_request('get', endpoint, *args, **kwargs)
def do_request(self, method, endpoint, **kwargs) -> requests.Response:
r = self.request(method, self.BASE_URL + endpoint, **kwargs)
r.raise_for_status()
return r
def get(self, endpoint: str, **kwargs):
return self.do_request('get', endpoint, **kwargs)
def post(self, endpoint: str, data=None, **kwargs):
return self.do_request('post', endpoint, data=data, **kwargs)
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment