Hello,
I use sqlalchemy to execute my tests on Postgres Database. I run like ~200 tests, every test uses its own setup and teardown. creates a connection and closes it after all.
I sometimes get ResourceClosedError ("StatementError: (sqlalchemy.exc.ResourceClosedError) This Connection is closed" in random places when tests are running one after another. It was not present on Sybase database. It seems like it wants to work on a closed connection...?
For a single test run I never got that.
Do you have any idea, what I could improve? This is a part of my code:
from sqlalchemy import create_engine
from sqlalchemy.orm import create_session
from configuration import PostgresConfiguration as PGS
class PostgresDatabase:
def __init__(self):
self.connection_string = r'postgresql://{}:{}@{}:{}/{}'.format(
PGS.DB_USER,
PGS.DB_PASSWORD,
PGS.DB_HOST.split(":")[0],
PGS.DB_HOST.split(":")[1],
PGS.DB_NAME)
self.engine = create_engine(self.connection_string, echo=True, label_length=30)
self.session = create_session(self.engine)
self.database_name = PGS.DB_NAME
def get_session(self):
"""
Retrieves current session.
:return: current session.
"""
return self.session
def close_session(self):
"""
Closes all open sessions.
"""
self.session.close_all()
def execute_query(self, query, commit=True):
"""
Executes query passes as a parameter.
:param query: passed query to be executed.
:param commit: if True session will be committed
:return query execution results
"""
if commit:
self.session.begin()
results = self.session.execute(query)
self.session.commit()
else:
results = self.session.execute(query)
return results