How to ensure that an empty table is populated with psycopg?

210 Views Asked by At

I am trying to find an efficient and reliable way to ensure that my INSERTs have indeed taken place for my empty tables.

While I would like to use a function like this

connection_info = "..."


def insert(statement: str, validate: bool):
    with psycopg.connect(connection_info) as connection:
        with connection.cursor() as cursor:
            cursor.execute(statement)
            if validate and not cursor.rowcount > 0:
               raise Error("It's not populated")
        connection.commit()

I don't know if there's a case where the cursor.rowcount is over zero, but not committed to the database. So, for now I use the following:

connection_info = "..."


def execute(statement: str):
    with psycopg.connect(connection_info) as connection:
        with connection.cursor() as cursor:
            cursor.execute(statement)
        connection.commit()


def is_populated(table: str) -> bool:
    with psycopg.connect(connection_info) as connection:
        with connection.cursor() as cursor:
            cursor.execute(f'SELECT 1 FROM {table}')
            rows = cursor.rowcount
        connection.commit()
    return rows > 0


def insert(statement: str, table: str, validate: bool):
    execute(statement)
    if validate and not is_populated(table):
        raise Error("it's not populated")

What's the best practice to ensure that my data are inserted into my tables?

PS I don't know if it makes and difference, but I'm using psycopg3

0

There are 0 best solutions below