I have some pytest project where I do some postgresql requests. I have some functions like get_json, get_user, get_message to get some specific objects from databases and I use these funcs inside of my fixtures. These funcs looks kinda like this:
def get_expected_status_code_API01(role, incident_type="a"):
connection = pg8000.connect(**constants.AUTOTESTS_DB_CREDS)
cursor = connection.cursor()
cursor.execute("SELECT status_code FROM pytest.api01_roles_and_status_codes WHERE role=%s AND incident_type=%s", [role, incident_type])
status_code = cursor.fetchall()[0][0]
cursor.close()
connection.close()
return status_code
So, yeah, I open and close connections in every query :( It was ok while I had not that much tests, but now I have an error "pg8000.dbapi.ProgrammingError: {'S': 'FATAL', 'V': 'FATAL', 'C': '53300', 'M': 'sorry, too many clients already', 'F': 'proc.c', 'L': '346', 'R': 'InitProcess'}". I supposed it would be ok since I created and closed connections, but for some reason it didn't work this way. I think I need some db sessions, but how exactly can I do that?