How to use _get_request_headers method in autotest

Best Python code snippet using autotest_python

api.py

Source:api.py

...5DEFAULT_TIMEOUT = 106class Api:7    def __init__(self, auth):8        self.auth = auth9    def _get_request_headers(self, headers):10        if headers is None:11            headers = {}12        token = self.auth.token13        if token:14            headers['Authorization'] = 'Bearer {}'.format(token)15        return headers16    def get(self, url, params=None, headers=None):17        headers = self._get_request_headers(headers)18        try:19            response = requests.get(url, params=params, headers=headers, timeout=DEFAULT_TIMEOUT)20            response.raise_for_status()21        except HTTPError as error:22            raise HarpokratHttpException(error, url)23        if response.status_code != 200:24            return None25        return response.json()26    def get_many(self, url, params=None, headers=None):27        return self.get(url, params=params, headers=headers)28    def post(self, url, data=None, params=None, headers=None):29        headers = self._get_request_headers(headers)30        try:31            response = requests.post(url, json=data, params=params, headers=headers, timeout=DEFAULT_TIMEOUT)32            response.raise_for_status()33        except HTTPError as error:34            raise HarpokratHttpException(error, url, data)35        if response.status_code != 200:36            return None37        return response.json()38    def patch(self, url, data=None, params=None, headers=None):39        headers = self._get_request_headers(headers)40        try:41            response = requests.patch(url, json=data, params=params, headers=headers, timeout=DEFAULT_TIMEOUT)42            response.raise_for_status()43        except HTTPError as error:44            raise HarpokratHttpException(error, url, data)45        if response.status_code != 200:46            return None47        return response.json()48    def put(self, url, data=None, params=None, headers=None):49        headers = self._get_request_headers(headers)50        try:51            response = requests.put(url, json=data, params=params, headers=headers, timeout=DEFAULT_TIMEOUT)52            response.raise_for_status()53        except HTTPError as error:54            raise HarpokratHttpException(error, url, data)55        if response.status_code != 200:56            return None57        return response.json()58    def delete(self, url, data=None, params=None, headers=None):59        headers = self._get_request_headers(headers)60        try:61            response = requests.delete(url, json=data, params=params, headers=headers, timeout=DEFAULT_TIMEOUT)62            response.raise_for_status()63        except HTTPError as error:64            raise HarpokratHttpException(error, url, data)65        if response.status_code != 200:66            return None...

sparql.py

Source:sparql.py

...6from SPARQLWrapper import SPARQLWrapper, JSON7import asyncio8import aiohttp9# Headers and extra query params were taken from the code of SPARQLQuery10def _get_request_headers():11    content_types = ["application/sparql-results+json", "application/json", "text/javascript", "application/javascript"]12    headers = {'Accept': ', '.join(content_types)}13    return headers14def _get_request_url(query):15    url = settings.WD_QUERY_ENDPOINT+query16    query_params = { 'query': query, 'format': 'json', 'output': 'json', 'result': 'json'}17    url = settings.WD_QUERY_ENDPOINT + '?' + urllib.parse.urlencode(query_params)18    return url19def query(query):20    """ Execute a SPARQL query synchronously """21    headers = _get_request_headers()22    url = _get_request_url(query)23    response = requests.get(url, headers=headers)24    text = response.text25    parsed = json.loads(text)26    return parsed27sparql_endpoint = SPARQLWrapper(settings.WD_QUERY_ENDPOINT)28def query_with_wrapper(query):29    # This is the old implementation using SPARQLWrapper.30    sparql_endpoint.setQuery(query)31    sparql_endpoint.setReturnFormat(JSON)32    run_query = sparql_endpoint.query()33    results = run_query.convert()34    return results35async def async_query(query):36    """ Asynchronously perform a SPARQL query """37    headers = _get_request_headers()38    url = _get_request_url(query)39    async with aiohttp.ClientSession() as session:40        async with session.get(url, headers=headers) as response:41            text = await response.content.read()42    43    parsed = json.loads(text)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.