i'm getting this error, would like a solution + the possible downside to using that solution, if applicable: in commit_results_worker
await asyncio.gather(*tasks_testing)
RuntimeError: cannot reuse already awaited coroutine

frmo this code:
import sys
from db_operations import *
import asyncio
import logging
import time
from datetime import datetime
from psycopg2 import pool # import pool module
from psycopg2.extras import execute_values

if sys.platform.startswith("win"):
selector = asyncio.WindowsSelectorEventLoopPolicy()
asyncio.set_event_loop_policy(selector)

logging.basicConfig(level=logging.DEBUG, format='%(asctime)s %(levelname)-8s %(message)s',
datefmt='%Y-%m-%d %H:%M:%S', handlers=[logging.FileHandler('proxy_checker.log'), logging.StreamHandler()])

logging.getLogger('asyncio').addFilter(lambda record: not (
record.levelname == 'ERROR' and 'An existing connection was forcibly closed by the remote host' in record.msg))

create a connection pool object with min and max connections
connection_pool = pool.ThreadedConnectionPool(1, 50, **db_config)

def get_proxies():
# get a connection from the pool
connection = connection_pool.getconn()
cursor = connection.cursor()

sql_more
Copy
select_proxy_query = '''
    WITH working_list_ids AS (
        SELECT
            list_id,
            COUNT(*) as working_count
        FROM
            conncheck
        WHERE
            status = 'working'
        GROUP BY
            list_id
        HAVING
            COUNT(*) >= 2
    ),
    ranked_working_proxies AS (
        SELECT
            list.id AS list_id,
            list.ip AS ip,
            list.port AS port,
            ROW_NUMBER() OVER (
                ORDER BY
                    working_list_ids.working_count,
                    list.id DESC
            ) as rn
        FROM
            list
            JOIN working_list_ids ON list.id = working_list_ids.list_id
    )
    SELECT
        list_id,
        ip,
        port
    FROM
        ranked_working_proxies
    WHERE
        rn BETWEEN 1 AND 500
'''
cursor.execute(select_proxy_query)
proxies = cursor.fetchall()

cursor.close()
# put back the connection to the pool
connection_pool.putconn(connection)

return proxies
def create_table_if_not_exists(table_name):
# get a connection from the pool
connection = connection_pool.getconn()
cursor = connection.cursor()

sql_more
Copy
create_table_query = f'''
    CREATE TABLE IF NOT EXISTS {table_name} (
        id SERIAL PRIMARY KEY,
        list_id INTEGER REFERENCES list(id),
        ctime TIMESTAMP NOT NULL,
        response_time FLOAT,
        status VARCHAR NOT NULL
    )
'''
cursor.execute(create_table_query)
connection.commit()

cursor.close()
# put back the connection to the pool
connection_pool.putconn(connection)
async def test_proxy_connectivity(sem, list_id, ip, port, index, results_queue):
async with sem:
start_time = time.time()
logging.debug(f'Testing proxy {index}: {ip}:{port}...')
try:
_, writer = await asyncio.wait_for(asyncio.open_connection(ip, port), timeout=20)
status = 'working'
writer.close()
await writer.wait_closed()
except asyncio.TimeoutError:
status = 'error_timeout'
except ConnectionRefusedError:
status = 'error_conn_refused'
except OSError as e:
status = f'error_os: {str(e)}'
except ConnectionResetError:
status = 'error_conn_reset'
except Exception as e:
status = f'error_other: {str(e)}'
response_time = time.time() - start_time

Copy
    logging.debug(f'Proxy {index}: {ip}:{port} is {status}')

    # Put the test result into the results_queue
    await results_queue.put((list_id, ip, port, status, response_time))
async def commit_results(sem, results_batch, connection_pool, table_name):
async with sem:
connection = connection_pool.getconn()
cursor = connection.cursor()

pgsql
Copy
    # Insert the results_batch into the database
    for result in results_batch:
        list_id, ip, port, status, response_time = result
        try:
            cursor.execute(
                f"INSERT INTO {table_name} (list_id, ctime, response_time, status) VALUES (%s, %s, %s, %s)",
                (list_id, datetime.now(), response_time, status)
            )
        except psycopg2.Error as e:
            print(f"Error inserting proxy data: {e}")

    connection.commit()
    cursor.close()
    connection_pool.putconn(connection)
    
async def commit_results_worker(sem_db_commit, results_queue, connection_pool, tasks_testing, table_name):
# Wait for all testing tasks to be done
await asyncio.gather(*tasks_testing)

basic
Copy
# Process the remaining results in results_queue
while not results_queue.empty():
    results = []
    batch_size = 100
    for _ in range(batch_size):
        result = await results_queue.get()
        if result is None:
            break
        results.append(result)

    await commit_results(sem_db_commit, results, connection_pool, table_name)
async def main():
# Create the tables
create_table_if_not_exists("conncheck")
create_table_if_not_exists("temp")

ini
Copy
proxies = get_proxies()
sem_proxy_testing = asyncio.Semaphore(500)
sem_db_commit = asyncio.Semaphore(5)

# Create an asynchronous queue for the test results
results_queue = asyncio.Queue()

# Create tasks for testing proxy connectivity
tasks_testing = [
    test_proxy_connectivity(sem_proxy_testing, list_id, ip, port, index + 1, results_queue)
    for index, (list_id, ip, port) in enumerate(proxies)
]

# Create a task for the commit results worker coroutine
task_commit_worker = asyncio.create_task(commit_results_worker(sem_db_commit, results_queue, connection_pool, tasks_testing, "temp"))

# Perform testing tasks and cancel the commit results worker tasks when done
await asyncio.gather(*tasks_testing)
await task_commit_worker

# close all the connections in the pool
connection_pool.closeall()
if name == "main":
logging.info('Starting proxy checker...')
start_time = time.time()
asyncio.run(main())
elapsed_time = time.time() - start_time
logging.info(f'Proxy checker finished in {elapsed_time:.2f} seconds.')