best practice of Psycopg
@app.post("/update_error_collection")
async def update_error_collection(request: ErrorUpdateRequest):
conn = get_db_connection()
cur = conn.cursor()
try:
cur.execute(
"INSERT INTO error_collection (input, output, user_id) VALUES (%s, %s, %s)",
(request.input_text, request.output_text, request.user_id)
)
conn.commit()
return {"message": "Error record added successfully"}
except Exception as e:
conn.rollback()
raise HTTPException(status_code=500, detail=f"Error updating error collection: {str(e)}")
finally:
cur.close()
conn.close()
def get_db_connection():
return psycopg2.connect(os.environ['DATABASE_URL'])
--------------------
Is this best practice? To individually obtain a connection and cursor each time and then close them.
async def update_error_collection(request: ErrorUpdateRequest):
conn = get_db_connection()
cur = conn.cursor()
try:
cur.execute(
"INSERT INTO error_collection (input, output, user_id) VALUES (%s, %s, %s)",
(request.input_text, request.output_text, request.user_id)
)
conn.commit()
return {"message": "Error record added successfully"}
except Exception as e:
conn.rollback()
raise HTTPException(status_code=500, detail=f"Error updating error collection: {str(e)}")
finally:
cur.close()
conn.close()
def get_db_connection():
return psycopg2.connect(os.environ['DATABASE_URL'])
--------------------
Is this best practice? To individually obtain a connection and cursor each time and then close them.
