46 lines
1003 B
Python
46 lines
1003 B
Python
import psycopg
|
|
from loguru import logger
|
|
|
|
def main():
|
|
try:
|
|
conn = psycopg.connect()
|
|
logger.info("Database connection opened")
|
|
except Exception as e:
|
|
logger.error(f"Error when opening database connection {e}"(
|
|
return
|
|
|
|
try:
|
|
with conn.cursor(cursor_factory=psycopg.extras.DictCursor) as cursor:
|
|
select_stmt = psycopg.sql.SQL("""
|
|
SELECT ...
|
|
""")
|
|
cursor.execute(select_stmt, (...))
|
|
row = cursor.fetchone()
|
|
|
|
if not row:
|
|
logger.error(f"No data found")
|
|
return
|
|
|
|
value = row["bla"]
|
|
|
|
# calculation
|
|
insert_stmt = psycopg.sql.SQL("""
|
|
INSERT ...
|
|
""")
|
|
cursor.execute(insert_stmt, (...))
|
|
conn.commit()
|
|
|
|
logger.info(f"Data successfully written to database")
|
|
except Exception as e:
|
|
logger.error(f"Error when selecting or writing database {e}")
|
|
conn.rollback()
|
|
finally:
|
|
conn.close()
|
|
logger.info("Database connection closed")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|
|
|
|
|