Created
February 15, 2023 23:19
-
-
Save Upgwades/79055cc87832699550efddee4c963f09 to your computer and use it in GitHub Desktop.
Sqlalchemy pytest convenience fixtures
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import pytest | |
| from models import Base | |
| from pytest_postgresql import factories | |
| from sqlalchemy import create_engine, text | |
| from sqlalchemy.orm import sessionmaker, scoped_session | |
| from sqlalchemy.pool import NullPool | |
| from sqlalchemy_utils import database_exists, create_database | |
| user = "postgres" | |
| host = "127.0.0.1" | |
| port = 5433 | |
| dbname = "test" | |
| connection = f"postgresql://{user}:@{host}:{port}/{dbname}" | |
| postgresql_proc = factories.postgresql_proc( | |
| user=user, | |
| host=host, | |
| port=port, | |
| dbname=dbname, | |
| ) | |
| @pytest.fixture(scope="session") | |
| def engine(postgresql_proc): | |
| yield create_engine(connection, poolclass=NullPool) | |
| @pytest.fixture(scope="session") | |
| def tables(engine): | |
| if not database_exists(engine.url): | |
| create_database(engine.url) | |
| Base.metadata.create_all(engine) | |
| yield | |
| Base.metadata.drop_all(engine) | |
| @pytest.fixture(scope="module") | |
| def session(engine, tables): | |
| Session = scoped_session(sessionmaker(bind=engine)) | |
| yield Session | |
| Session.close() | |
| @pytest.fixture(scope="module") | |
| def cleanup(session): | |
| yield | |
| for table in Base.metadata.sorted_tables: | |
| session.execute(text(f'ALTER TABLE "{table.name}" DISABLE TRIGGER ALL;')) | |
| session.execute(table.delete()) | |
| session.execute(text(f'ALTER TABLE "{table.name}" ENABLE TRIGGER ALL;')) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment