Python-сообщество

 mport requests
class Test_new_location():
    """Работа с новой локацией"""
    def test_create_new_location(self):
        """Создание новой локации"""
        base_url = "https://rahulshettyacademy.com"          # базовая url
        key = "?key=qaclick123"             # Параметр для всех запросов (из документации)
        """Создание новой локации"""
        post_resourse = "/maps/api/place/add/json"           #  Ресурс метода POST
        post_url = base_url + post_resourse + key
        print(post_url)
    #
        json_for_create_new_location = {
            "location": {
                "lat": -38.383494,
                "lng": 33.427362
            }, "accuracy": 50,
            "name": "Frontline house",
            "phone_number": "(+91) 983 893 3937",
            "address": "29, side layout, cohen 09",
            "types": [
                "shoe park",
                "shop"
            ],
            "website": "http://google.com",
            "language": "French-IN"
        }
        result_post = requests.post(post_url, json = json_for_create_new_location)
        print("Статус-код : " + str(result_post.status_code))
        check_post = result_post.json()
        place_id = check_post.get("place_id")
        print("Place_id : " + place_id)
        for i in range(1, 6):
            print(result_post.text)
            place = open('post.txt', 'a')
            place.write(place_id)
            place.write(" \n")
            place.close()
        """Проверка создания новой локации"""
        get_resourse = "/maps/api/place/get/json"
        fw = open('post.txt', 'r')
        text = fw.read()
        fw.close()
        print(text)
        get_url = base_url + get_resourse + key + "&place_id=" + text
        print(get_url)
        result_get = requests.get(get_url)
        print(result_get.text)
        print("Статус-код : " + str(result_get.status_code))
        assert 200 == result_get.status_code
        if result_get.status_code == 200:
            print("Успешно!!! Проверка создания новой локации прошла успешно")
        else:
            print("Провал!!! Запрос ошибочный")
test = Test_new_location()
test.test_create_new_location()

  
import asyncio
import time
 
import aiohttp
import aiofiles
 
base_url = r"https://rahulshettyacademy.com"
key = r"?key=qaclick123"
post_resourse = "/maps/api/place/add/json"
get_resourse = "/maps/api/place/get/json"
del_resourse = '/maps/api/place/delete/json'
 
async def get_place_id(session):
    print('post')
    json_for_create_new_location = {
        "location": {
            "lat": -38.383494,
            "lng": 33.427362
        }, "accuracy": 50,
        "name": "Frontline house",
        "phone_number": "(+91) 983 893 3937",
        "address": "29, side layout, cohen 09",
        "types": [
            "shoe park",
            "shop"
        ],
        "website": "http://google.com",
        "language": "French-IN"
    }
    url = base_url + post_resourse + key
    async with session.post(url, json=json_for_create_new_location) as response:
        if response.ok:
            data = await response.json()
            if data['status'] == "OK":
                return data
 
async def validate_place_id(session, place_id):
    print('get')
    url = base_url + get_resourse + key + "&place_id=" + place_id
    async with session.get(url) as response:
        if response.ok:
            return place_id
 
async def delete_place_id(session, place_id):
    print('del')
    json = {'place_id': place_id}
    url = base_url + del_resourse + key
    async with session.delete(url, json=json) as response:
        if response.ok:
            data = await response.json()
            return data
 
async def main():
    async with aiohttp.ClientSession() as session:
        post_tasks = [get_place_id(session) for i in range(5)]
        async with aiofiles.open('place_id.txt', 'w') as f:
            for cor in asyncio.as_completed(post_tasks):
                data = await cor
                if data:
                    await f.write(f"{data['place_id']}\n")
                    print('write')
                else:
                    print('error')
 
        async with aiofiles.open('place_id.txt', 'r') as f:
            place_id_lst = [place_id.strip() for place_id in await f.readlines()]
            get_tasks = [
                validate_place_id(session, place_id) for place_id in place_id_lst
            ]
            get_result = await asyncio.gather(*get_tasks)
            print(get_result)
            if all(get_result):
                print('location create all')
            del_result = await asyncio.gather(delete_place_id(session, place_id_lst[1]),
                                              delete_place_id(session, place_id_lst[3]))
            print(del_result)
 
            async with aiofiles.open('place_id_2.txt', 'w') as wf:
                get_tasks = [
                    validate_place_id(session, place_id) for place_id in place_id_lst
                ]
                for cor in asyncio.as_completed(get_tasks):
                    data = await cor
                    if data:
                        await wf.write(f"{data}\n")
 
if __name__ == '__main__':
    s = time.time()
    asyncio.run(main())
    print(f'time {time.time() - s:.2f} s')