KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run yandex-tank automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
