Function main_sync

tests/scripts/copytest.py:34–46

(args: Namespace)

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

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


async def main_async(args: Namespace) -> None: