I am trying to find an efficient and reliable way to ensure that my INSERT
s 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