Function main_async

tests/scripts/copytest.py:49–61

(args: Namespace)

async def main_async(args: Namespace) -> None:
    test = CopyPutTest(args)
    async with await psycopg.AsyncConnection.connect(args.dsn) as conn:
        async with conn.cursor() as cur:
            await cur.execute(test.get_table_stmt())
            writer = getattr(psycopg.copy, args.writer)(cur) if args.writer else None
            t0 = time()
            async with cur.copy(test.get_copy_stmt(), writer=writer) as copy:
                for i in range(args.nrecs):
                    await copy.write_row(test.get_record())
            tf = time()

    logger.info("time to copy: %.6f sec", tf - t0)


class CopyPutTest: