TestMu 2025 - The Largest Virtual Software Testing Conference is Back | 50,000+ Attendees

How to use _create_headers method in localstack

Best Python code snippet using localstack_python

mpower.py

...6from kitiwa.settings import MPOWER_TOKEN, MPOWER_MASTER_KEY,\7    MPOWER_PRIVATE_KEY, MPOWER_ENDPOINT_OPR_TOKEN_REQUEST,\8    MPOWER_ENDPOINT_OPR_TOKEN_CHARGE, MPOWER_ENDPOINT_CHECK_INVOICE_STATUS,\9    MPOWER_RESPONSE_SUCCESS, MPOWER_STATUS_COMPLETED10def _create_headers():11    return {12        'content-type': 'application/json',13        'MP-Master-Key': MPOWER_MASTER_KEY,14        'MP-Private-Key': MPOWER_PRIVATE_KEY,15        'MP-Token': MPOWER_TOKEN16    }17def opr_token_request(amount, mpower_phone_number,18                      invoice_desc='', store_name='Kitiwa'):19    # convert to local phone number format20    mpower_phone_number = '0{}'.format(mpower_phone_number[4::])21    payload = {22        'invoice_data': {23            'invoice': {24                'total_amount': amount,25                'description': invoice_desc26            },27            'store': {28                'name': store_name29            }30        },31        'opr_data': {32            'account_alias': mpower_phone_number33        }34    }35    headers = _create_headers()36    try:37        response = requests.post(38            MPOWER_ENDPOINT_OPR_TOKEN_REQUEST,39            data=json.dumps(payload),40            headers=headers41        )42        decoded_response = response.json()43        response_code = decoded_response['response_code']44        response_text = decoded_response['response_text']45        opr_token = decoded_response['token']46        invoice_token = decoded_response['invoice_token']47    except KeyError as e:48        opr_token = invoice_token = 'N/A'49    except RequestException as e:50        message = 'ERROR - MPOWER (opr_token_request for no: {}): {}'51        log_error(message.format(mpower_phone_number, repr(e)))52        response_code = opr_token = invoice_token = 'N/A'53        response_text = repr(e)54    return response_code, response_text, opr_token, invoice_token55def opr_charge_action(opr_token, confirm_token):56    headers = _create_headers()57    payload = {58        'token': opr_token,59        'confirm_token': confirm_token60    }61    try:62        response = requests.post(63            MPOWER_ENDPOINT_OPR_TOKEN_CHARGE,64            data=json.dumps(payload),65            headers=headers66        )67        decoded_response = response.json()68        response_code = decoded_response['response_code']69        response_text = decoded_response['response_text']70    except (RequestException, KeyError) as e:71        message = 'ERROR - MPOWER (opr_charge_action for token {}): {}'72        log_error(message.format(opr_token, repr(e)))73        response_code = "N/A"74        response_text = repr(e)75    return response_code, response_text76def check_invoice_status(invoice_token):77    headers = _create_headers()78    endpoint = MPOWER_ENDPOINT_CHECK_INVOICE_STATUS.format(invoice_token)79    try:80        response = requests.get(81            endpoint,82            headers=headers83        )84        decoded_response = response.json()85        if decoded_response['response_code'] != MPOWER_RESPONSE_SUCCESS:86            message = 'ERROR - MPOWER (check_invoice_status for token {}): transaction not found'87            raise MPowerException88        if decoded_response['status'] != MPOWER_STATUS_COMPLETED:89            message = 'ERROR - MPOWER (check_invoice_status for token {}): transaction not completed'90            raise MPowerException91        return True...

twitter_api.py

Source: twitter_api.py

...14        if tweet_fields is not None:15            url += f"&tweet.fields={tweet_fields}"16        if since_id is not None:17            url += f"&since_id={since_id}"18        headers = TwitterApi._create_headers()19        response = requests.request("GET", url, headers=headers)20        if response.status_code != 200:21            raise Exception(response.status_code, response.text)22        return response.json()23    @staticmethod24    def _create_headers() -> dict:25        """26        Create authorization header with bearer token27        :return: Authorization header with bearer token28        """29        return {"Authorization": f"Bearer {twitter_config.api['bearer_token']}"}30    @staticmethod31    def get_rules() -> any:32        """33        Get rules34        :return: Rules35        """36        response = requests.get(37            "https://api.twitter.com/2/tweets/search/stream/rules", headers=TwitterApi._create_headers()38        )39        if response.status_code != 200:40            raise Exception(f"Cannot get rules (HTTP {response.status_code}): {response.text}")41        return response.json()42    @staticmethod43    def delete_rules(rules) -> any:44        """45        Delete given rules46        :param rules: Rules to be deleted47        """48        if rules is None or "data" not in rules:49            return None50        ids = list(map(lambda rule: rule["id"], rules["data"]))51        payload = {"delete": {"ids": ids}}52        response = requests.post(53            "https://api.twitter.com/2/tweets/search/stream/rules",54            headers=TwitterApi._create_headers(),55            json=payload56        )57        if response.status_code != 200:58            raise Exception(f"Cannot delete rules (HTTP {response.status_code}): {response.text}")59        return response.json()60    @staticmethod61    def set_rules(rules: list) -> any:62        """63        Set rules64        :param rules: Rules to be set65        sample_rules = [66            {"value": "dog has:images", "tag": "dog pictures"},67            {"value": "cat has:images -grumpy", "tag": "cat pictures"},68        ]69        """70        # You can adjust the rules if needed71        payload = {"add": rules}72        response = requests.post(73            "https://api.twitter.com/2/tweets/search/stream/rules",74            headers=TwitterApi._create_headers(),75            json=payload,76        )77        if response.status_code != 201:78            raise Exception(f"Cannot add rules (HTTP {response.status_code}): {response.text}")79        return response.json()80    @staticmethod81    def get_stream() -> any:82        response = requests.get(83            "https://api.twitter.com/2/tweets/search/stream", headers=TwitterApi._create_headers(), stream=True,84        )85        if response.status_code != 200:86            raise Exception(f"Cannot get stream (HTTP {response.status_code}): {response.text}")...

apiclient.py

Source: apiclient.py

...13        if query:14            params = urlencode(query)15            return self.prefix_url + endpoint + "?" + params16        return self.prefix_url + endpoint17    def _create_headers(self, extra: dict = None) -> dict:18        headers = deepcopy(self.auth_header)19        if extra:20            headers.update(extra)21        return headers22    def fetch_events(self, query: dict) -> requests.Response:23        url = self._create_url("api/v1/events/", query)24        headers = self._create_headers()25        response = requests.get(url=url, headers=headers)26        logger.info(response.text)27        return response28    def fetch_event_progresses(self, query: dict) -> requests.Response:29        url = self._create_url("api/v1/events/progresses/", query)30        headers = self._create_headers()31        response = requests.get(url=url, headers=headers)32        logger.info(response.text)33        return response34    def create_and_update_event_progresses(self, query: dict) -> requests.Response:35        url = self._create_url("api/v1/events/progresses/")36        headers = self._create_headers(37            {38                "Content-Type": "application/json",39            }40        )41        response = requests.post(url=url, headers=headers, data=json.dumps(query))42        logger.info(response.text)43        return response44    def delete_event_progresses(self, query: dict) -> requests.Response:45        url = self._create_url("api/v1/events/progresses/", query)46        headers = self._create_headers()47        response = requests.delete(url=url, headers=headers)48        logger.info(response.text)49        return response50apiClient = ApiClient(51    prefix_url=WMS_API_BASE_URL,52    auth_header={"Authorization": f"Basic {WMS_API_BASIC_AUTHORIZATION}"},...

Blogs

Check out the latest blogs from LambdaTest on this topic:

What will come after “agile”?

I think that probably most development teams describe themselves as being “agile” and probably most development teams have standups, and meetings called retrospectives.There is also a lot of discussion about “agile”, much written about “agile”, and there are many presentations about “agile”. A question that is often asked is what comes after “agile”? Many testers work in “agile” teams so this question matters to us.

Guide To Find Index Of Element In List with Python Selenium

In an ideal world, you can test your web application in the same test environment and return the same results every time. The reality can be difficult sometimes when you have flaky tests, which may be due to the complexity of the web elements you are trying to perform an action on your test case.

[LambdaTest Spartans Panel Discussion]: What Changed For Testing & QA Community And What Lies Ahead

The rapid shift in the use of technology has impacted testing and quality assurance significantly, especially around the cloud adoption of agile development methodologies. With this, the increasing importance of quality and automation testing has risen enough to deliver quality work.

Get A Seamless Digital Experience With #LambdaTestYourBusiness????

The holidays are just around the corner, and with Christmas and New Year celebrations coming up, everyone is busy preparing for the festivities! And during this busy time of year, LambdaTest also prepped something special for our beloved developers and testers – #LambdaTestYourBusiness

Failures may be inevitable, but that doesn’t prevent success

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.