How to use __make_api_request method in yandex-tank

Best Python code snippet using yandex-tank

skyetel.py

Source:skyetel.py

...13        self.__url = urls.URLs()14        self.__session = requests.Session()15        self.__session.headers = {'X-AUTH-SID': x_auth_sid, 'X-AUTH-SECRET': x_auth_secret}16    @limits(calls=120, period=60)17    def __make_api_request(self, request_type, endpoint, data=None, json=None):18        try:19            if request_type == 'GET':20                response = self.__session.get(endpoint, data=data, json=json)21            elif request_type == 'POST':22                response = self.__session.post(endpoint, data=data, json=json)23            elif request_type == 'PATCH':24                response = self.__session.patch(endpoint, data=data, json=json)25            elif request_type == 'DELETE':26                response = self.__session.delete(endpoint, data=data, json=json)27            else:28                raise errors.ValidationError('Invalid Request Type')29            if response.status_code != 200:30                content = response.json()31                raise errors.APIError(content['ERROR'])32        except (requests.ConnectionError, requests.Timeout) as e:33            raise errors.Unavailable('API Unavailable: {}'.format(e)) from None34        return response.json()35    def get_audio_recordings_list(self, items_per_page=10, page_offset=0, query=None, search=None, sort=None):36        """37            Get a list of the phone call recordings.38        :param items_per_page: integer, defaults to 10 records returned per request39        :param page_offset: integer, combines with items_per_page40        :param query: string, wildcard search on all string fields41        :param search: dict, format 'field':'query'42        :param sort: list[string], list of fields to sort, prefix a '-' for descending sort43        :return: list[AudioRecording], List of AudioRecording objects44        """45        parameters = '?page[limit]={}&page[offset]={}'.format(items_per_page, page_offset)46        if query:47            parameters += '&filter[query]={}'.format(query)48        if search:49            for field in search:50                parameters += '&filter[{}]={}'.format(field, search[field])51        if sort:52            parameters += '&sort={}'.format(sort[0])53            for x in range(1, len(sort)):54                parameters += ',{}'.format(sort[x])55        response = self.__make_api_request('GET', self.__url.audio_recordings_url() + parameters)56        if response:57            for x in range(0, len(response)):58                response[x]['insert_time'] = datetime.strptime(response[x]['insert_time'], SKYETEL_DATESTRING)59                response[x]['start_time'] = datetime.strptime(response[x]['start_time'], SKYETEL_DATESTRING)60                response[x]['org'] = responses.Organization(**response[x]['org'])61                response[x] = responses.AudioRecording(**response[x])62        return response63    def get_audio_recording_url(self, recording_id):64        """65            Get the URL for the audio file of a specific call recording66        :param recording_id: integer, ID of a call recording from recording list67        :return: string, URL of call recording audio file68        """69        response = self.__make_api_request('GET', self.__url.audio_recording_download_url(recording_id))70        return response.get('download_url', None)71    def get_audio_transcriptions_list(self, items_per_page=10, page_offset=0, query=None, search=None, sort=None):72        """73            Get a list of all phone call transcriptions74        :param items_per_page: integer, defaults to 10 records returned per request75        :param page_offset: integer, combines with items_per_page76        :param query: string, wildcard search on all string fields77        :param search: dict, format 'field':'query'78        :param sort: list[string], list of fields to sort, prefix a '-' for descending sort79        :return: list[AudioTranscription], list of AudioTranscription objects80        """81        parameters = '?page[limit]={}&page[offset]={}'.format(items_per_page, page_offset)82        if query:83            parameters += '&filter[query]={}'.format(query)84        if search:85            for field in search:86                parameters += '&filter[{}]={}'.format(field, search[field])87        if sort:88            parameters += '&sort={}'.format(sort[0])89            for x in range(1, len(sort)):90                parameters += ',{}'.format(sort[x])91        response = self.__make_api_request('GET', self.__url.audio_transcriptions_url() + parameters)92        if response:93            for x in range(0, len(response)):94                response[x]['insert_time'] = datetime.strptime(response[x]['insert_time'], SKYETEL_DATESTRING)95                response[x]['start_time'] = datetime.strptime(response[x]['start_time'], SKYETEL_DATESTRING)96                response[x]['org'] = responses.Organization(**response[x]['org'])97                response[x] = responses.AudioTranscription(**response[x])98        return response99    def get_audio_transcription_url(self, transcription_id):100        """101            Get the URL for the text log of a specific call transcription102        :param transcription_id: integer, ID of a call transcription from transcription list103        :return: string, URL of call transcription text log104        """105        response = self.__make_api_request('GET', self.__url.audio_transcription_download_url(transcription_id))106        return response.get('download_url', None)107    def get_audio_transcription_text(self, transcription_id):108        """109            Get the text log for a specific call transcription110        :param transcription_id: integer, ID of a call transcription from transcription list111        :return: dict, dictionary of both parties text and timestamps112        """113        response = self.__make_api_request('GET', self.__url.audio_transcription_download_url(transcription_id))114        url = response.get('download_url', '')115        if url:116            transcript = requests.get(url).json()117            return transcript118        else:119            return None120    def get_billing_balance(self):121        """122            Get the remaining balance on the account123        :return: float, billing balance remaining on the account124        """125        response = self.__make_api_request('GET', self.__url.balance_url())126        return float(response.get('BALANCE', 0))127    def get_organization_statement(self, year=None, month=None):128        """129            Get the monthly statement for the entire organization account. Defaults to current month.130        :param year: integer, four digit year of the statement month131        :param month: integer, single or double digit, with January corresponding to 1132        :return: BillingStatement133        """134        parameters = '?'135        if year:136            parameters += 'year={}'.format(year)137            if month:138                parameters += '&'139        if month:140            parameters += 'month={}'.format(month)141        response = self.__make_api_request('GET', self.__url.organization_statement_url() + parameters)142        if response:143            if response['transactions']:144                for x in range(0, len(response['transactions'])):145                    response['transactions'][x]['transaction_date'] = datetime.strptime(146                        response['transactions'][x]['transaction_date'], SKYETEL_DATESTRING)147                    response['transactions'][x] = responses.StatementTransaction(**response['transactions'][x])148            if response['taxes']:149                for x in range(0, len(response['taxes'])):150                    response['taxes'][x] = responses.StatementTax(**response['taxes'][x])151            if response['statement']:152                response['statement']['totals']['phone_numbers'] = responses.PhoneNumberTotals(153                    **response['statement']['totals']['phone_numbers'])154                response['statement'] = responses.StatementTotals(**response['statement']['totals'])155            response = responses.BillingStatement(**response)156        return response157    def get_endpoints_list(self, items_per_page=10, page_offset=0):158        """159            Get list of SIP Endpoints160        :param items_per_page: integer, defaults to 10 records returned per request161        :param page_offset: integer, combines with items_per_page162        :return: list[Endpoint], list of Endpoint objects163        """164        parameters = '?page[limit]={}&page[offset]={}'.format(items_per_page, page_offset)165        response = self.__make_api_request('GET', self.__url.endpoints_url() + parameters)166        if response:167            for x in range(0, len(response)):168                response[x]['endpoint_group'] = responses.EndpointGroup(**response[x]['endpoint_group'])169                response[x]['org'] = responses.Organization(org_id=response[x]['org']['id'],170                                                            org_name=response[x]['org']['name'])171                response[x] = responses.Endpoint(**response[x])172        return response173    def create_endpoint(self, ip, priority, description, endpoint_group_id, endpoint_group_name, port=5060,174                        transport="udp"):175        """176            Create a new SIP Endpoint in an Endpoint Group177        :param ip: string, IPv4 Address of the Endpoint178        :param priority: integer, the higher the number, the higher the priority179        :param description: string, short note to describe the Endpoint180        :param endpoint_group_id: integer, Endpoint Group ID181        :param endpoint_group_name: string, Endpoint Group Name182        :param port: integer, SIP Port Number for the Endpoint (default 5060)183        :param transport: string, SIP Transport method, UDP (default) or TCP184        :return: Endpoint, object representation of the created Endpoint185        """186        parameters = {'ip': ip, 'port': port, 'transport': transport, 'priority': priority, 'description': description,187                      'endpoint_group_id': endpoint_group_id, 'endpoint_group_name': endpoint_group_name}188        response = self.__make_api_request('POST', self.__url.endpoints_url(), data=parameters)189        if response:190            response['endpoint_group'] = responses.EndpointGroup(**response['endpoint_group'])191            response['org'] = responses.Organization(org_id=response['org']['id'],192                                                     org_name=response['org']['name'])193            response = responses.Endpoint(**response)194        return response195    def update_endpoint(self, endpoint_id, ip, priority, description, endpoint_group_id, endpoint_group_name, port=5060,196                        transport="udp"):197        """198            Update an existing Endpoint's details199        :param endpoint_id: integer, assigned ID for the Endpoint200        :param ip: string, IPv4 Address of the Endpoint201        :param priority: integer, the higher the number, the higher the priority202        :param description: string, short note to describe the Endpoint203        :param endpoint_group_id: integer, Endpoint Group ID204        :param endpoint_group_name: string, Endpoint Group Name205        :param port: integer, SIP Port Number for the Endpoint (default 5060)206        :param transport: string, SIP Transport method, UDP (default) or TCP207        :return: Endpoint, object representation of the updated Endpoint208        """209        parameters = {'ip': ip, 'port': port, 'transport': transport, 'priority': priority, 'description': description,210                      'endpoint_group_id': endpoint_group_id, 'endpoint_group_name': endpoint_group_name}211        response = self.__make_api_request('PATCH', self.__url.endpoint_url(endpoint_id), data=parameters)212        if response:213            response['endpoint_group'] = responses.EndpointGroup(**response['endpoint_group'])214            response['org'] = responses.Organization(org_id=response['org']['id'],215                                                     org_name=response['org']['name'])216            response = responses.Endpoint(**response)217        return response218    def get_phonenumber_e911(self, phonenumber_id):219        """220            Get the E911 address associated with a phone number221        :param phonenumber_id: integer, assigned ID for the Phone Number222        :return: E911Address, object representation of the associated E911 Address223        """224        response = self.__make_api_request('GET', self.__url.phonenumber_e911address_url(phonenumber_id))225        if response:226            response = responses.E911Address(**response)227        return response228    def create_phonenumber_e911(self, phonenumber_id, caller_name, address1, address2, community, state, postal_code):229        """230            Create an E911 address for a phone number and enable E911 processing231        :param phonenumber_id: integer, assigned ID for the Phone Number232        :param caller_name: string, Name associated with the Phone Number233        :param address1: string, First Address Line for Emergency Response234        :param address2: string, Second Address Line for Emergency Response235        :param community: string, City for Emergency Response236        :param state: string, Two letter State Abbreviation for Emergency Response237        :param postal_code: string, Postal Code for Emergency Response238        :return: E911Address, object representation of the associated E911 Address239        """240        parameters = {'caller_name': caller_name, 'address1': address1, 'address2': address2, 'community': community,241                      'state': state, 'postal_code': postal_code}242        response = self.__make_api_request('POST', self.__url.phonenumber_e911address_url(phonenumber_id),243                                           data=parameters)244        if response:245            response = responses.E911Address(**response)246        return response247    def update_phonenumber_e911(self, phonenumber_id: int, caller_name, address1, address2, community, state,248                                postal_code):249        """250            Update an E911 address for a phone number251        :param phonenumber_id: integer, assigned ID for the Phone Number252        :param caller_name: string, Name associated with the Phone Number253        :param address1: string, First Address Line for Emergency Response254        :param address2: string, Second Address Line for Emergency Response255        :param community: string, City for Emergency Response256        :param state: string, Two letter State Abbreviation for Emergency Response257        :param postal_code: string, Postal Code for Emergency Response258        :return: E911Address, object representation of the associated E911 Address259        """260        parameters = {'caller_name': caller_name, 'address1': address1, 'address2': address2, 'community': community,261                      'state': state, 'postal_code': postal_code}262        response = self.__make_api_request('PATCH', self.__url.phonenumber_e911address_url(phonenumber_id),263                                           data=parameters)264        if response:265            response = responses.E911Address(**response)266        return response267    def get_phonenumbers(self, items_per_page=10, page_offset=0, query: str = None, search: Dict = None,268                         sort: List = None):269        """270            Get a list of all Phone Numbers associated with the organization account271        :param items_per_page: integer, defaults to 10 records returned per request272        :param page_offset: integer, combines with items_per_page273        :param query: string, wildcard search on all string fields274        :param search: dict, format 'field':'query'275        :param sort: list[string], list of fields to sort, prefix a '-' for descending sort276        :return: list[PhoneNumber], list of PhoneNumber objects277        """278        parameters = '?page[limit]={}&page[offset]={}'.format(items_per_page, page_offset)279        if query:280            parameters += '&filter[query]={}'.format(query)281        if search:282            for field in search:283                parameters += '&filter[{}]={}'.format(field, search[field])284        if sort:285            parameters += '&sort={}'.format(sort[0])286            for x in range(1, len(sort)):287                parameters += ',{}'.format(sort[x])288        response = self.__make_api_request('GET', self.__url.phonenumbers_url() + parameters)289        if response:290            for x in range(0, len(response)):291                response[x]['number'] = int(response[x]['number'])292                if response[x]['forward']:293                    response[x]['forward'] = int(response[x]['forward'])294                if response[x]['failover']:295                    response[x]['failover'] = int(response[x]['failover'])296                if response[x]['endpoint_group']:297                    response[x]['endpoint_group'] = responses.EndpointGroup(**response[x]['endpoint_group'])298                if response[x]['tenant']:299                    response[x]['tenant'] = responses.Tenant(**response[x]['tenant'])300                if response[x]['origination']:301                    response[x]['origination'] = responses.Origination(**response[x]['origination'])302                if response[x]['e911address']:303                    response[x]['e911address'] = responses.E911Address(**response[x]['e911address'])304                response[x]['intl_balance'] = float(response[x]['intl_balance'])305                response[x]['intl_reserve'] = float(response[x]['intl_reserve'])306                response[x]['org']['account_number'] = int(response[x]['org']['account_number'])307                response[x]['org']['support_pin'] = int(response[x]['org']['support_pin'])308                response[x]['org']['balance'] = float(response[x]['org']['balance'])309                response[x]['org']['auto_recharge_reserve'] = float(response[x]['org']['auto_recharge_reserve'])310                response[x]['org'] = responses.ExtendedOrganization(**response[x]['org'])311                response[x] = responses.PhoneNumber(**response[x])312        return response313    def create_off_network_phonenumber(self, number: str):314        """315            Creates an Off-Network Phone Number316        :param number: string, 11 digit phone number to register317        :return: OffNetworkPhoneNumber, object representation of the Off-Network Phone Number318        """319        parameters = {'number': str(number)}320        response = self.__make_api_request('POST', self.__url.phonenumbers_offnetwork_url(), json=parameters)321        if response:322            response = responses.OffNetworkPhoneNumber(**response)323        return response324    def update_phonenumber(self, phonenumber_id: int, update_data: responses.PhoneNumberUpdate):325        """326            Update features and settings of a Phone Number327        :param phonenumber_id: integer, assigned Id for the Phone Number328        :param update_data: PhoneNumberUpdate, object containing changes to apply to the Phone Number329        :return: PhoneNumberUpdate, object representation of the Phone Number's features and settings330        """331        data = update_data.as_dict()332        response = self.__make_api_request('PATCH', self.__url.phonenumber_url(phonenumber_id), data=data)333        if response:334            response = responses.PhoneNumberUpdate(**response)335        return response336    def get_available_phonenumbers(self, search_filter: responses.PhoneNumberFilter = None):337        """338            Get a list of Phone Numbers available for purchase, with filtering. Phone Numbers are held server-side339            for 10 minutes340        :param search_filter: PhoneNumberFilter, object containing filters to apply to search341        :return: list[string], List of Phone Numbers available for purchase342        """343        params = ''344        if search_filter:345            params = search_filter.params()346        response = self.__make_api_request('GET', self.__url.phonenumbers_ordersearch_url() + params)347        return response348    def get_rate_centers(self, state: str = None):349        """350            Get a list of Rate Centers in a given State351        :param state: string, Two letter State Abbreviation352        :return: list[RateCenter], list of RateCenter objects353        """354        params = ''355        if state:356            params = '?state={}'.format(state)357        response = self.__make_api_request('GET', self.__url.phonenumbers_ratecenters_url() + params)358        if response:359            for x in range(0, len(response)):360                response[x] = responses.RateCenter(**response[x])361        return response362    def order_phonenumbers(self, number_list: List[responses.NumberPurchase]):363        """364            UNTESTED: Order Phone Numbers365        :param number_list: list[NumberPurchase], List of NumberPurchase objects with associated MOU366        :return: list[PhoneNumberUpdate], List of PhoneNumberUpdate objects367        """368        data = {}369        for num in number_list:370            data['numbers[{}][mou]'.format(num.number)] = num.mou371        response = self.__make_api_request('POST', self.__url.phonenumbers_order_url(), data=data)372        if response:373            for x in range(0, len(response)):374                response[x] = responses.PhoneNumberUpdate(**response[x])375        return response376    def get_local_phonunumbers_count(self):377        """378            Get a count of local Phone Numbers in the organization379        :return: int, Count of local Phone Numbers380        """381        response = self.__make_api_request('GET', self.__url.phonenumbers_localcount_url())382        return int(response.get('TOTAL', 0))383    def get_tollfree_phonenumbers_count(self):384        """385            Get a count of Toll-Free Phone Numbers in the organization386        :return: int, Count of Toll-Free Phone Numbers387        """388        response = self.__make_api_request('GET', self.__url.phonenumbers_tfcount_url())389        return int(response.get('TOTAL', 0))390    def get_sms_receipts(self, items_per_page=10, page_offset=0, query: str = None, search: Dict = None,391                         sort: List = None):392        """393            Get a list of received SMS/MMS messages394        :param items_per_page: integer, defaults to 10 records returned per request395        :param page_offset: integer, combines with items_per_page396        :param query: string, wildcard search on all string fields397        :param search: dict, format 'field':'query'398        :param sort: list[string], list of fields to sort, prefix a '-' for descending sort399        :return: list[SMSMessage], list of SMSMessage objects400        """401        parameters = '?page[limit]={}&page[offset]={}'.format(items_per_page, page_offset)402        if query:403            parameters += '&filter[query]={}'.format(query)404        if search:405            for field in search:406                parameters += '&filter[{}]={}'.format(field, search[field])407        if sort:408            parameters += '&sort={}'.format(sort[0])409            for x in range(1, len(sort)):410                parameters += ',{}'.format(sort[x])411        response = self.__make_api_request('GET', self.__url.smsreceipts_url() + parameters)412        if response:413            for x in range(0, len(response)):414                response[x]['time'] = datetime.strptime(response[x]['time'], SKYETEL_DATESTRING)415                response[x]['cost'] = float(response[x]['cost'])416                response[x]['org']['account_number'] = int(response[x]['org']['account_number'])417                response[x]['org']['support_pin'] = int(response[x]['org']['support_pin'])418                response[x]['org']['balance'] = float(response[x]['org']['balance'])419                response[x]['org']['auto_recharge_reserve'] = float(response[x]['org']['auto_recharge_reserve'])420                response[x]['org'] = responses.ExtendedOrganization(**response[x]['org'])421                response[x] = responses.SMSMessage(**response[x])422        return response423    def get_endpoint_health(self, items_per_page: int = 10, page_offset: int = 0):424        """425            Get a list of all Endpoints and their associated health status426        :param items_per_page: integer, defaults to 10 records returned per request427        :param page_offset: integer, combines with items_per_page428        :return: list[EndpointHealth], list of EndpointHealth objects429        """430        parameters = '?page[limit]={}&page[offset]={}'.format(items_per_page, page_offset)431        response = self.__make_api_request('GET', self.__url.endpoint_health_url() + parameters)432        if response:433            for x in range(0, len(response)):434                response[x] = responses.EndpointHealth(**response[x])435        return response436    def get_daily_traffic_counts(self, items_per_page: int = 10, page_offset: int = 0, start_time_min: datetime = None,437                                 start_time_max: datetime = None, tz_string: str = None):438        """439            Get a list of Traffic Counts per-day between two specified dates440        :param items_per_page: integer, defaults to 10 records returned per request441        :param page_offset: integer, combines with items_per_page442        :param start_time_min: datetime, filters data after this date443        :param start_time_max: datetime, filters data before this date444        :param tz_string: string, standard Time Zone string (ex. America/New_York)445        :return: list[TrafficCount], list of TrafficCount objects446        """447        parameters = '?page[limit]={}&page[offset]={}'.format(items_per_page, page_offset)448        if start_time_min:449            parameters += '&start_time_min={}'.format(start_time_min.strftime(SKYETEL_DATESTRING))450        if start_time_max:451            parameters += '&start_time_max={}'.format(start_time_max.strftime(SKYETEL_DATESTRING))452        if tz_string:453            parameters += '&tz={}'.format(tz_string)454        response = self.__make_api_request('GET', self.__url.traffic_count_url() + parameters)455        if response:456            for x in range(0, len(response)):457                response[x]['date'] = datetime.strptime(response[x]['date'], SKYETEL_DATESTRING)458                response[x] = responses.TrafficCount(**response[x])459        return response460    def get_daily_traffic_channels(self, items_per_page: int = 10, page_offset: int = 0,461                                   start_time_min: datetime = None,462                                   start_time_max: datetime = None, tz_string: str = None):463        """464            Get a list of Channels used per-day between two specified dates465        :param items_per_page: integer, defaults to 10 records returned per request466        :param page_offset: integer, combines with items_per_page467        :param start_time_min: datetime, filters data after this date468        :param start_time_max: datetime, filters data before this date469        :param tz_string: string, standard Time Zone string (ex. America/New_York)470        :return: list[ChannelCount], list of ChannelCount objects471        """472        parameters = '?page[limit]={}&page[offset]={}'.format(items_per_page, page_offset)473        if start_time_min:474            parameters += '&start_time_min={}'.format(start_time_min.strftime(SKYETEL_DATESTRING))475        if start_time_max:476            parameters += '&start_time_max={}'.format(start_time_max.strftime(SKYETEL_DATESTRING))477        if tz_string:478            parameters += '&tz={}'.format(tz_string)479        response = self.__make_api_request('GET', self.__url.channel_count_url() + parameters)480        if response:481            for x in range(0, len(response)):482                response[x]['date'] = datetime.strptime(response[x]['date'], SKYETEL_DATESTRING)483                response[x]['channel_count'] = int(response[x]['channel_count'])484                response[x] = responses.ChannelCount(**response[x])485        return response486    def get_hourly_call_count(self, items_per_page: int = 10, page_offset: int = 0,487                              start_time_min: datetime = None, start_time_max: datetime = None, tz_string: str = None):488        """489            Get a list of Calls placed per-hour between two specified dates490        :param items_per_page: integer, defaults to 10 records returned per request491        :param page_offset: integer, combines with items_per_page492        :param start_time_min: datetime, filters data after this date493        :param start_time_max: datetime, filters data before this date494        :param tz_string: string, standard Time Zone string (ex. America/New_York)495        :return: list[CallCount], list of CallCount objects496        """497        parameters = '?page[limit]={}&page[offset]={}'.format(items_per_page, page_offset)498        if start_time_min:499            parameters += '&start_time_min={}'.format(start_time_min.strftime(SKYETEL_DATESTRING))500        if start_time_max:501            parameters += '&start_time_max={}'.format(start_time_max.strftime(SKYETEL_DATESTRING))502        if tz_string:503            parameters += '&tz={}'.format(tz_string)504        response = self.__make_api_request('GET', self.__url.traffic_hourly_url() + parameters)505        if response:506            for x in range(0, len(response)):507                response[x]['date'] = datetime.strptime(response[x]['date'], SKYETEL_TIMESTRING)508                response[x]['call_count'] = int(response[x]['call_count'])509                response[x] = responses.CallCount(**response[x])510        return response511    def get_tenant_statements(self, year=None, month=None):512        """513            Get all Tenant Statements in a given month. Defaults to current month514        :param year: integer, four digit year of the statement month515        :param month: integer, single or double digit, with January corresponding to 1516        :return: list[TenantStatement], list of TenantStatement objects517        """518        parameters = '?'519        if year:520            parameters += 'year={}'.format(year)521            if month:522                parameters += '&'523        if month:524            parameters += 'month={}'.format(month)525        response = self.__make_api_request('GET', self.__url.tenant_statements_url() + parameters)526        if response:527            for x in range(0, len(response)):528                response[x]['month'] = datetime.strptime(response[x]['month'], SKYETEL_DATESTRING)529                response[x]['org'] = responses.Organization(org_id=response[x]['org']['id'],530                                                            org_name=response[x]['org']['org_name'])531                response[x]['tenant'] = responses.Tenant(**response[x]['tenant'])532                response[x]['fields']['totals']['phone_numbers'] = \533                    responses.TenantPhoneNumberTotals(**response[x]['fields']['totals']['phone_numbers'])534                response[x]['fields']['totals'] = responses.TenantStatementTotals(**response[x]['fields']['totals'])535                response[x] = responses.TenantStatement(id=response[x]['id'], month=response[x]['month'],536                                                        org=response[x]['org'], tenant=response[x]['tenant'],537                                                        totals=response[x]['fields']['totals'])538        return response539    def get_tenant_invoices(self):540        """541            UNTESTED: Get all Tenant Invoices542        :return: list[TenantInvoice], list of TenantInvoice objects543        """544        response = self.__make_api_request('GET', self.__url.tenant_invoices_url())545        if response:546            for x in range(0, len(response)):547                response[x]['scheduled_date'] = datetime.strptime(response[x]['scheduled_date'], SKYETEL_TIMESTRING)548                response[x] = responses.TenantInvoice(**response[x])549        return response550    def create_onetime_tenant_invoice(self, tenant_id: int, billing: responses.TenantBillingProfile):551        """552            UNTESTED: Create a One-Time Tenant Invoice from a Tenant Billing Profile553        :param tenant_id: integer, ID of the Tenant for the Invoice554        :param billing: TenantBillingProfile, a TenantBillingProfile object555        :return: string, Stripe Invoice ID556        """557        data = asdict(billing)558        for x in range(0, len(data.get('billing_products', []))):559            data['billing_products'][x] = asdict(data['billing_products'][x])560        response = self.__make_api_request('POST', self.__url.tenant_invoice_url(tenant_id), data=data)561        return response.get('stripe_invoice_id', None)562    def void_tenant_invoice(self, tenant_id: int):563        """564            UNTESTED: Void an Invoice for a Tenant565        :param tenant_id: integer, ID of the Tenant566        :return: None567        """568        self.__make_api_request('DELETE', self.__url.tenant_invoice_url(tenant_id))569        return None570    def get_billing_products(self):571        """572            UNTESTED: Get a list of Tenant Billing Products573        :return: list[TenantBillingProduct], List of TenantBillingProduct objects574        """575        response = self.__make_api_request('GET', self.__url.tenant_products_url())576        if response:577            for x in range(0, len(response)):578                response[x] = responses.TenantBillingProduct(**response[x])579        return response580    def get_tenants(self, items_per_page=10, page_offset=0, query: str = None, search: Dict = None, sort: List = None):581        """582            Get a list of Tenants583        :param items_per_page: integer, defaults to 10 records returned per request584        :param page_offset: integer, combines with items_per_page585        :param query: string, wildcard search on all string fields586        :param search: dict, format 'field':'query'587        :param sort: list[string], list of fields to sort, prefix a '-' for descending sort588        :return: list[ExtendedTenant], list of ExtendedTenant objects589        """590        parameters = '?page[limit]={}&page[offset]={}'.format(items_per_page, page_offset)591        if query:592            parameters += '&filter[query]={}'.format(query)593        if search:594            for field in search:595                parameters += '&filter[{}]={}'.format(field, search[field])596        if sort:597            parameters += '&sort={}'.format(sort[0])598            for x in range(1, len(sort)):599                parameters += ',{}'.format(sort[x])600        response = self.__make_api_request('GET', self.__url.tenants_url()+parameters)601        if response:602            for x in range(0, len(response)):603                response[x]['phonenumber'] = int(response[x]['phonenumber'])604                if response[x]['billing_profile']['billing_products']:605                    for y in range(0, len(response[x]['billing_profile']['billing_products'])):606                        response[x]['billing_profile']['billing_products'][x] = responses.TenantInvoiceProduct(607                            **response[x]['billing_profile']['billing_products'][x])608                response[x]['billing_profile'] = responses.TenantBillingProfile(**response[x]['billing_profile'])609                response[x]['contact_phonenumber'] = int(response[x]['contact_phonenumber'])610                response[x]['date_added'] = datetime.strptime(response[x]['date_added'], SKYETEL_DATESTRING)611                response[x]['org'] = responses.Organization(org_id=response[x]['org']['id'],612                                                            org_name=response[x]['org']['org_name'])613                response[x] = responses.ExtendedTenant(**response[x])614        return response

datadog_client.py

Source:datadog_client.py

...44        with open(self.default_config_file, "w+") as new_config:45            yaml.dump(config, new_config, default_flow_style=False)46        print(f"ð¾ Created config file at {self.default_config_file}")47        return config48    def __make_api_request(self, resource, method, key, *args, **kwargs):49        """50        Make a request to the Datadog api.51        Raises:52            RequestForbiddenError53        """54        try:55            objs = getattr(getattr(api, resource), method)(*args, **kwargs)56        except Exception as ex:57            str_resp = ex.args[0]58            if "403 Forbidden" in str_resp:59                raise RequestForbiddenError(60                    f"Have your credentials been set in {self.default_config_file}?"61                )62        else:63            to_return = objs64            if key:65                to_return = objs[key]66            return sorted(to_return, key=lambda x: x.get('title', x.get('name')))67        return []68    def get_dashboards(self) -> list:69        return self.__make_api_request("Dashboard", "get_all", "dashboards")70    def get_dashboard_detail(self, dashboard_id):71        return self.__make_api_request("Dashboard", "get", None, dashboard_id)72    def update_dashboard(self, dashboard_id, **kwargs):73        return self.__make_api_request(74            "Dashboard", "update", None, dashboard_id, **kwargs75        )76    def get_monitors(self):77        return self.__make_api_request("Monitor", "get_all", None)78    def update_monitor(self, monitor_id, **kwargs):79        return self.__make_api_request("Monitor", "update", None, monitor_id, **kwargs)80    def configure(self, api_key, app_key):81        self._create_config_file(api_key, app_key)82@dataclass83class Dashboard:84    emoji: str = "ð"85    name = 'dashboard'86    key = 'dashboards'87@dataclass88class Monitor:89    emoji: str = "â ï¸"90    name = 'monitor'91    key = 'monitors'92DATADOG_OBJECTS = {93    Monitor.key: Monitor,...

slack.py

Source:slack.py

...30        }31      }]),32      'attachments': kwargs.get('attachments', [])33    }34    return self.__make_api_request(targetUrl, payload)35  def post_ephemeral_message(self, text, target_user_id, **kwargs):36    targetUrl = kwargs.get('response_url', self.__get_api_url("chat.postEphemeral"))37    payload = {38      'channel': kwargs.get('channel_id', self.CHANNEL_ID),39      'text': text,40      'as_user': kwargs.get('as_user', False),41      'blocks': kwargs.get('blocks', [{42        "type": "section",43        "text": {44          "type": "mrkdwn",45          "text": text46        }47      }]),48      'attachments': kwargs.get('attachments', []),49      'user': target_user_id50    }51    return self.__make_api_request(targetUrl, payload)52  def update_message(self, text, target_ts, **kwargs):53    targetUrl = self.__get_api_url("chat.update")54    payload = {55      'channel': kwargs.get('channel_id', self.CHANNEL_ID),56      'text': text,57      'blocks': kwargs.get('blocks', [{58        "type": "section",59        "text": {60          "type": "mrkdwn",61          "text": text62        }63      }]),64      'as_user': kwargs.get('as_user', False),65      'ts': target_ts66    }67    if kwargs.get('attachments', None):68      payload['attachments'] = kwargs.get('attachments')69    if kwargs.get('blocks', None):70      payload['blocks'] = kwargs.get('blocks')71    return self.__make_api_request(targetUrl, payload)72  def get_user_info(self, user_id, **kwargs):73    targetUrl = self.__get_api_url("users.info")74    payload = {75      'user': user_id,76      'include_locale': kwargs.get('include_local', True)77    }78    return self.__make_api_request(targetUrl, payload, True)79  def __make_api_request(self, target_url, payload, url_form_encoded=False):80    if isinstance(payload, dict):81      if url_form_encoded:82        payload = urllib.parse.urlencode(payload)83      else:84        payload = json.dumps(payload)85    elif not isinstance(payload, str):86      raise TypeError("Expected json string or dict")87    if url_form_encoded:88      headers={'Content-Type': "application/x-www-form-urlencoded"}89    else:90      headers={'Content-Type': "application/json"}91    headers['Authorization'] = "Bearer {}".format(self.API_TOKEN)92    print(headers)93    r = requests.post(target_url,...

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.