Задать вопрос
  • Как отключить перенос строк при форматировании HTML в VSCODE?

    @3FANG
    «Alt+Z» или через пункт главного меню «Вид – Перенос по словам»
    Ответ написан
    Комментировать
  • Как итерировать асинхронный генератор, который уже находится в работе?

    @3FANG Автор вопроса
    Для решения своей проблемы я воспользовался очередью(asyncio.Queue), т.е. я стал пользоваться не асинхронным генератором, а корутиной, которая вместо возврата значения (с помощью yield) добавляла его в очередь. А в другую корутину я уже передавал не асинхронный генератор, а эту самую очередь и потом из нее доставал значение.
    Вот что вышло:
    import asyncio
    from time import time
    
    
    PROXY_KEYS = {
        "http://1.1.1.1:9404": "ff4e67941cb08be2fbc164fbbd00",
        "http://1.1.1.1:9309": "9286f4c4de633ed5fd39420sff1d",
        "http://1.1.1.1:9551": "4c62aa15fa9743e55a1ff1e0696e",
    }
    
    
    async def get_proxy_and_key(proxy_and_key: dict, queue: asyncio.Queue):
        """
        Получить пару прокси-APIключ.
    
        Проходится циклом по всем парам, каждую он отдает на 10(9) запросов,
        если перед повторным использованием пары не прошла 1 секунда, то скрипт ожидает
        истечения этого времени для избежания ошибки Ratelimit exceed.
        """
        while True:
            t0 = time()
            for proxy, key in proxy_and_key.items():
                for _ in range(9):
                    await queue.put((proxy, key))
            end = round(time() - t0, 2)
            if end < 1:
                # можно сделать ожидание истечения оставшегося до 1 секунды времени,
                # но я поставил именно 1 сек, чтоб наверняка 
                await asyncio.sleep(1)
    
    
    async def request(queue: asyncio.Queue):
        while True:
            pair = await queue.get()
            # Какие-то манипуляции, но мы просто будем выводить на печать
            print(pair)
    
    
    async def main():
        queue = asyncio.Queue(maxsize=100)
        queue_task = asyncio.create_task(get_proxy_and_key(PROXY_KEYS, queue))
    
        tasks = []
        for i in range(100):
            task = asyncio.create_task(request(queue))
            tasks.append(task)
    
        await asyncio.gather(queue_task, *tasks)
    
    
    asyncio.run(main())
    Ответ написан
    Комментировать
  • Почему SimpleTestCase.assertContains возвращает False?

    @3FANG Автор вопроса
    Я нашел решение проблемы.
    Я не установил во второй функции в метод, выполняющий запрос - self.client.post() аргумент follow=True.
    Т.е. у меня сохранялся только первый адрес и состояние - ('/tasks/6/delete/', 302), а не все промежуточные адреса и коды состояния.
    Я правильно понял механизм assertContains - он проходит по всем полученным в ходе запроса адресам и состояниям и проверяет, что response произвел данный status_code и что text появляется в его content. И если он получил хоть одно совпадение, то возвращает True?
    Ответ написан
    Комментировать
  • Как объединить форму и ListView?

    @3FANG
    Попробуй UserPlugs наследовать от forms.ModelForm, а не от forms.Form. Соответственно, убери из AccountSettings метод get_context_data(). В форме замени цикл на {{ form }}. Чтобы данные оставались в форме, определи get_initial() в AccountSettings
    Ответ написан
    Комментировать