diff --git a/README.md b/README.md
index 4b419c4..fe4c841 100644
--- a/README.md
+++ b/README.md
@@ -29,33 +29,36 @@ It automatically provisions a dedicated test database per worker and runs Alembi
* `sqlalchemy` >= 2.0
* `alembic` >= 1.16
* `pytest` >= 8.4
+
+
+
* `pytest-xdist` >= 3.0 (Optional)
* `pytest-asyncio` >= 1.3.0 (Optional)
## Supported dialects
-| Driver | Status |
-| ------ | ------ |
-| `sqlite+pysqlite` | ✅ Supported
-| `sqlite+pysqlcipher` | ✅ Supported
-| `postgresql+psycopg` | ✅ Supported
-| `postgresql+psycopg2` | ✅ Supported
-| `postgresql+pg8000` | ✅ Supported
-| `postgresql+psycopg2cffi` | ✅ Supported
-| `postgresql+asyncpg` | ✅ Supported
-| `mariadb+mysqldb` | ✅ Supported
-| `mariadb+pymysql` | ✅ Supported
+| Driver | Status
+| -------------------------- | ------
+| `sqlite+pysqlite` | ✅ Supported
+| `sqlite+pysqlcipher` | ✅ Supported
+| `postgresql+psycopg` | ✅ Supported
+| `postgresql+psycopg2` | ✅ Supported
+| `postgresql+pg8000` | ✅ Supported
+| `postgresql+psycopg2cffi` | ✅ Supported
+| `postgresql+asyncpg` | ✅ Supported
+| `mariadb+mysqldb` | ✅ Supported
+| `mariadb+pymysql` | ✅ Supported
| `mariadb+mariadbconnector` | ✅ Supported
-| `mariadb+asyncmy` | ✅ Supported
-| `mariadb+aiomysql` | ✅ Supported
-| `mariadb+cymysql` | ✅ Supported
-| `mysql+mysqldb` | ✅ Supported
-| `mysql+pymysql` | ✅ Supported
-| `mysql+mysqlconnector` | ✅ Supported
-| `mysql+asyncmy` | ✅ Supported
-| `mysql+aiomysql` | ✅ Supported
-| `mysql+cymysql` | ✅ Supported
+| `mariadb+asyncmy` | ✅ Supported
+| `mariadb+aiomysql` | ✅ Supported
+| `mariadb+cymysql` | ✅ Supported
+| `mysql+mysqldb` | ✅ Supported
+| `mysql+pymysql` | ✅ Supported
+| `mysql+mysqlconnector` | ✅ Supported
+| `mysql+asyncmy` | ✅ Supported
+| `mysql+aiomysql` | ✅ Supported
+| `mysql+cymysql` | ✅ Supported
> [!NOTE]
> If you need an implementation for your particular SQLAlchemy driver, please consider contributing to this project.
@@ -66,6 +69,11 @@ It automatically provisions a dedicated test database per worker and runs Alembi
```bash
pip install pytest-sqlalchemy-alembic
```
+or
+```bash
+pip install pytest-sqlalchemy-alembic[xdist] # with pytest-xdist
+pip install pytest-sqlalchemy-alembic[async] # with pytest-asyncio
+```
## Minimal configuration
@@ -77,16 +85,22 @@ Configuration could be done in `pyproject.toml`, `pytest.ini` or via pytest fixt
```toml
[tool.pytest.ini_options]
-sqlalchemy_session_maker = "example_project.db:SessionLocal"
+sqlalchemy_alembic_configs = [
+ {session_maker = "example_project_async.db:AsyncSessionLocal"},
+]
```
+In TOML, [array of tables](https://toml.io/en/v1.1.0#array) should be used to provide config values
### `pytest.ini`
```ini
[pytest]
-sqlalchemy_session_maker = example_project.db:SessionLocal
+sqlalchemy_alembic_configs =
+ {"session_maker": "example_project.db:SessionLocal"}
```
+In INI each line should be a valid JSON object
+
### `conftest.py`
```python
@@ -95,8 +109,14 @@ from pytest_sqlalchemy_alembic.config import PluginConfig
from example_project.db import SessionLocal
@pytest.fixture(scope='session')
-def sqlalchemy_alembic_plugin_config() -> PluginConfig:
- return PluginConfig.build(session_maker=SessionLocal)
+def sqlalchemy_alembic_plugin_configs(pytestconfig: pytest.Config) -> list[PluginConfig]:
+ return [
+ PluginConfig.build(
+ session_maker=SessionLocal,
+ create_db=pytestconfig.getoption('create_db'),
+ no_migrations=pytestconfig.getoption('no_migrations'),
+ )
+ ]
```
@@ -104,51 +124,31 @@ def sqlalchemy_alembic_plugin_config() -> PluginConfig:
### Config file options
-| Option | Description
-| ------ | -----------
-| `sqlalchemy_session_maker` | Import path to a sessionmaker instance
-| `sqlalchemy_metadata` | Import path to SQLAlchemy metadata. Usually `metadata` attribute of a declarative base class.
Used for non-alembic database schema population based on metadata. Metadata is extracted from the alembic config if this option is empty
-| `sqlalchemy_engine` | Import path to SQLAlchemy engine instance.
If empty, created using `engine_url` and `engine_kwargs` or extracted from the sessionmaker instance
-| `sqlalchemy_engine_url` | SQLAlchemy engine URL. Extracted from `engine` if empty
-| `sqlalchemy_engine_kwargs` | Import path to a dict containing engine kwargs
-| `sqlalchemy_orm_loader` | Import path to module or callable that loads all ORM necessary models
-| `sqlalchemy_engine_scope` | Defines at which scope test engine should be activated (`session` or `function`).
Default: `session`
+| Option | Description
+| -------------------- | -----------
+| `session_maker` | Import path to a sessionmaker instance
+| `metadata` | Import path to SQLAlchemy metadata. Usually `metadata` attribute of a declarative base class.
Used for non-alembic database schema population based on metadata. Metadata is extracted from the alembic config if this option is empty
+| `engine` | Import path to SQLAlchemy engine instance.
If empty, created using `engine_url` and `engine_kwargs` or extracted from the sessionmaker instance
+| `engine_url` | SQLAlchemy engine URL. Extracted from `engine` if empty
+| `engine_kwargs` | Import path to a dict containing engine kwargs
+| `orm_loader` | Import path to module or callable that loads all ORM necessary models
+| `scope` | Defines at which scope test engine should be activated (`session` or `function`).
Default: `session`
+| `skip_db_management` | If this entry should skip test database management.
Useful in case of multiple engines, which use the same database.
Default: `false`
### Fixture override options
-Arguments of `PluginConfig.build` class method to use in the `sqlalchemy_alembic_plugin_config` fixture.
-
-| Argument | Type | Description
-| -------- | ---- | -----------
-| `session_maker` | `sa.orm.sessionmaker[Session]` | Instance of a sessionmaker
-| `metadata` | `sa.MetaData \| Sequence[sa.MetaData]` | SQLAlchemy metadata
-| `engine` | `sa.Engine` | Sqlalchemy engine instance
-| `engine_url` | `str` | SQLAlchemy engine URL
-| `engine_kwargs` | `dict[str, Any]` | `dict` with kwargs for `sa.create_engine` function. E.g. `{'json_serializer': my_json_serializer}`
-| `orm_loader` | `Callable[[], Any]` | Callable, that will load all ORM necessary models
-| `engine_scope` | `Literal['session', 'function']` | Defines at which scope test engine should be activated
-
-
-## Combining file configuration and fixture override
-
-In this example `session_maker` will be defined in `pyproject.toml` and `metadata` will be directly imported in the `conftest.py`.
+Arguments of `PluginConfig.build` class method to use in the `sqlalchemy_alembic_plugin_configs` fixture.
-```toml
-# pyproject.toml
-[tool.pytest.ini_options]
-sqlalchemy_session_maker = "example_project.db:SessionLocal"
-```
-
-```python
-# conftest.py
-import pytest
-from pytest_sqlalchemy_alembic.config import PluginConfig
-from example_project.models import Base
-
-@pytest.fixture(scope='session')
-def sqlalchemy_alembic_plugin_config(pytestconfig: pytest.Config) -> PluginConfig:
- return PluginConfig.build(config=pytestconfig, metadata=Base.metadata)
-```
+| Argument | Type | Description
+| -------------------- | -------------------------------------- | -----------
+| `session_maker` | `sa.orm.sessionmaker[Session]` | Instance of a sessionmaker
+| `metadata` | `sa.MetaData \| Sequence[sa.MetaData]` | SQLAlchemy metadata
+| `engine` | `sa.Engine` | Sqlalchemy engine instance
+| `engine_url` | `str` | SQLAlchemy engine URL
+| `engine_kwargs` | `dict[str, Any]` | `dict` with kwargs for `sa.create_engine` function. E.g. `{'json_serializer': my_json_serializer}`
+| `orm_loader` | `Callable[[], Any]` | Callable, that will load all ORM necessary models
+| `scope` | `Literal['session', 'function']` | Defines at which scope test engine should be activated
+| `skip_db_management` | `bool` | If this entry should skip test database management.
Useful in case of multiple engines, which use the same database
## Usage
@@ -177,16 +177,16 @@ def test_my_service():
## Pytest run flags
-| Flag | Description |
-| ---- | ----------- |
+| Flag | Description
+| ---- | -----------
| `--createdb` / `--create-db` | Drop and recreate the test database before the session
| `--nomigrations` / `--no-migrations` | Skip Alembic migrations and use `metadata.create_all()` instead
## Fixtures
-| Fixture | Scope | Output | Description |
-| ------- | ----- | ------ | ----------- |
-| `sqlalchemy_alembic_plugin_config` | session | This plugin's config | Extension point to override config values from python context
-| `sqlalchemy_alembic_setup_session` | session | Test SQLAlchemy engine | Session-scoped fixture to set up test database
-| `sqlalchemy_alembic_setup_session` | function | Test SQLAlchemy engine or `None` | Function-scoped fixture to set up test database.
Used only in case of `engine_scope='function'`
+| Fixture | Scope | Output | Description
+| ----------------------------------- | -------- | ----------------------------------------------- | -----------
+| `sqlalchemy_alembic_plugin_configs` | session | List of this plugin configs | Extension point to override config values from python context
+| `sqlalchemy_alembic_setup_session` | session | Sequence of test SQLAlchemy engines | Session-scoped fixture to set up test database
+| `sqlalchemy_alembic_setup_function` | function | Sequence of test SQLAlchemy engines or `None`'s | Function-scoped fixture to set up test database.
Used only in case of `scope='function'`
diff --git a/example_project/pyproject.toml b/example_project/pyproject.toml
index b8c8f54..a2da1d3 100644
--- a/example_project/pyproject.toml
+++ b/example_project/pyproject.toml
@@ -49,9 +49,6 @@ reinstall-package = ["pytest-sqlalchemy-alembic"]
[tool.uv.sources]
pytest-sqlalchemy-alembic = {path = "../"}
-[tool.pytest.ini_options]
-sqlalchemy_session_maker = "example_project.db:SessionLocal"
-
[tool.mypy]
check_untyped_defs = true
extra_checks = true
diff --git a/example_project/pytest.ini b/example_project/pytest.ini
new file mode 100644
index 0000000..9e76202
--- /dev/null
+++ b/example_project/pytest.ini
@@ -0,0 +1,4 @@
+[pytest]
+filterwarnings = error
+sqlalchemy_alembic_configs =
+ {"session_maker": "example_project.db:SessionLocal"}
diff --git a/example_project_async/pyproject.toml b/example_project_async/pyproject.toml
index 7b5d763..2c5572d 100644
--- a/example_project_async/pyproject.toml
+++ b/example_project_async/pyproject.toml
@@ -13,20 +13,20 @@ requires-python = ">=3.10"
version = "0.1.0"
[project.optional-dependencies]
+mariadb = [
+ "aiomysql>=0.3.2",
+ "asyncmy>=0.2.11",
+]
+mysql = [
+ "aiomysql>=0.3.2",
+ "asyncmy>=0.2.11",
+]
postgres = [
"asyncpg>=0.31.0",
]
sqlite = [
"aiosqlite>=0.22.1",
]
-mariadb = [
- "aiomysql>=0.3.2",
- "asyncmy>=0.2.11",
-]
-mysql = [
- "aiomysql>=0.3.2",
- "asyncmy>=0.2.11",
-]
[dependency-groups]
dev = [
@@ -60,8 +60,9 @@ warn_unused_ignores = true
[tool.pytest.ini_options]
asyncio_mode = "auto"
filterwarnings = ["error"]
-sqlalchemy_session_maker = "example_project_async.db:AsyncSessionLocal"
-asyncio_default_fixture_loop_scope = "function"
+sqlalchemy_alembic_configs = [
+ {session_maker = "example_project_async.db:AsyncSessionLocal"},
+]
[tool.alembic]
file_template = "%%(year)d-%%(month).2d-%%(day).2d_%%(rev)s_%%(slug)s"
diff --git a/src/pytest_sqlalchemy_alembic/config.py b/src/pytest_sqlalchemy_alembic/config.py
index 0f4f1eb..bb5a22e 100644
--- a/src/pytest_sqlalchemy_alembic/config.py
+++ b/src/pytest_sqlalchemy_alembic/config.py
@@ -1,5 +1,9 @@
+from __future__ import annotations
+
+import json
from collections.abc import Callable, Sequence
-from typing import Any, Literal, TypeVar
+from dataclasses import dataclass
+from typing import Any, Literal, TypeAlias, TypeGuard
import pytest
import sqlalchemy as sa
@@ -8,9 +12,10 @@
from typing_extensions import Self
from .dialects import DATABASE_MANAGERS, BaseDatabaseManager
-from .utils import get_alembic_target_metadata, import_string
+from .utils import TypeCheckDataclass, get_alembic_target_metadata, import_string
-SM = TypeVar('SM', async_sessionmaker[AsyncSession], sessionmaker[Session])
+AnySessionMaker: TypeAlias = async_sessionmaker[AsyncSession] | sessionmaker[Session]
+AnyEngine: TypeAlias = AsyncEngine | sa.Engine
class ConfigValidationError(ValueError):
@@ -18,55 +23,85 @@ class ConfigValidationError(ValueError):
class PluginConfig:
- __slots__ = ('database_manager', 'engine', 'engine_kwargs', 'engine_scope', 'metadata', 'session_maker')
+ __slots__ = ('create_db', 'database_manager', 'engine', 'engine_kwargs', 'metadata', 'no_migrations', 'scope', 'session_maker', 'skip_db_management')
def __init__(
self,
- session_maker: async_sessionmaker[AsyncSession] | sessionmaker[Session] | None,
+ *,
+ session_maker: AnySessionMaker | None,
engine_kwargs: dict[str, Any],
- engine: sa.Engine | AsyncEngine,
+ engine: AnyEngine,
database_manager: type[BaseDatabaseManager],
metadata: Sequence[sa.MetaData],
- engine_scope: Literal['session', 'function'],
+ scope: Literal['session', 'function'],
+ skip_db_management: bool = False,
+ create_db: bool = False,
+ no_migrations: bool = False,
):
self.session_maker = session_maker
self.engine_kwargs = engine_kwargs
self.engine = engine
self.database_manager = database_manager
self.metadata = metadata
- self.engine_scope = engine_scope
+ self.scope = scope
+ self.skip_db_management = skip_db_management
+ self.create_db = create_db
+ self.no_migrations = no_migrations
+
+ @classmethod
+ def build_list(cls, config: pytest.Config) -> list[Self]:
+ raw_value: list[str] | list[Any] = config.getini('sqlalchemy_alembic_configs')
+ create_db = config.getoption('create_db')
+ no_migrations = config.getoption('no_migrations')
+
+ configs = []
+ for raw_entry in raw_value:
+ if (entry := IniEntry.parse(raw_entry)) is not None:
+ configs.append(cls.build(ini_entry=entry, create_db=create_db, no_migrations=no_migrations))
+ return configs
@classmethod
def build(
cls,
- config: pytest.Config | None = None,
- session_maker: SM | None = None,
+ *,
+ ini_entry: IniEntry | None = None,
+ session_maker: AnySessionMaker | None = None,
metadata: sa.MetaData | Sequence[sa.MetaData] | None = None,
- engine: sa.Engine | AsyncEngine | None = None,
+ engine: AnyEngine | None = None,
engine_url: str | None = None,
engine_kwargs: dict[str, Any] | None = None,
orm_loader: Callable[[], None] | None = None,
- engine_scope: Literal['session', 'function'] = 'session',
+ scope: Literal['session', 'function'] | None = None,
+ skip_db_management: bool | None = None,
+ create_db: bool = False,
+ no_migrations: bool = False,
) -> Self:
- session_maker = cls._parse_session_maker(config, session_maker)
- engine_kwargs = cls._parse_engine_kwargs(config, engine_kwargs)
- engine = cls._parse_engine(config, engine, engine_url, engine_kwargs, session_maker)
- metadata = cls._parse_metadata(config, metadata)
- cls._load_orm_models(config, orm_loader)
- engine_scope = cls._parse_engine_scope(config, engine_scope)
+ if ini_entry is None:
+ ini_entry = IniEntry()
+
+ session_maker = cls._parse_session_maker(session_maker, ini_entry.session_maker)
+ engine_kwargs = cls._parse_engine_kwargs(engine_kwargs, ini_entry.engine_kwargs)
+ engine = cls._parse_engine(engine, ini_entry.engine, engine_url, ini_entry.engine_url, engine_kwargs, session_maker)
+ metadata = cls._parse_metadata(metadata, ini_entry.metadata)
+ scope = cls._parse_scope(scope, ini_entry.scope)
+ skip_db_management = ini_entry.skip_db_management if skip_db_management is None else skip_db_management
if engine.url.drivername not in DATABASE_MANAGERS:
msg = f'Unsupported database dialect: {engine.url.drivername}'
raise ConfigValidationError(msg)
database_manager = DATABASE_MANAGERS[engine.url.drivername]
+ cls._load_orm_models(orm_loader, ini_entry.orm_loader)
return cls(
session_maker=session_maker,
engine_kwargs=engine_kwargs,
engine=engine,
database_manager=database_manager,
metadata=metadata,
- engine_scope=engine_scope,
+ scope=scope,
+ skip_db_management=skip_db_management,
+ create_db=create_db,
+ no_migrations=no_migrations,
)
@classmethod
@@ -78,86 +113,86 @@ def import_from_string(cls, import_path: str) -> Any:
raise ConfigValidationError(msg) from e
@classmethod
- def _parse_session_maker(cls, config: pytest.Config | None, session_maker: SM | None) -> SM | None:
+ def _parse_session_maker(cls, session_maker: AnySessionMaker | None, session_maker_path: str | None) -> AnySessionMaker | None:
if session_maker is None:
- if config is None:
+ if session_maker_path is None:
return None
- session_maker_path = config.getini('sqlalchemy_session_maker')
if not session_maker_path:
return None
session_maker = cls.import_from_string(session_maker_path)
if not isinstance(session_maker, (sessionmaker, async_sessionmaker)):
- msg = f'`sqlalchemy_session_maker` must be a sessionmaker or async_sessionmaker, got {type(session_maker).__name__!r}'
+ msg = f'`session_maker` must be a sessionmaker or async_sessionmaker, got {type(session_maker).__name__!r}'
raise ConfigValidationError(msg)
return session_maker
@classmethod
- def _parse_engine_kwargs(cls, config: pytest.Config | None, engine_kwargs: dict[str, Any] | None) -> dict[str, Any]:
+ def _parse_engine_kwargs(cls, engine_kwargs: dict[str, Any] | None, engine_kwargs_path: str | None) -> dict[str, Any]:
if engine_kwargs is None:
- if config is None:
+ if engine_kwargs_path is None:
return {}
- engine_kwargs_path = config.getini('sqlalchemy_engine_kwargs')
if engine_kwargs_path:
engine_kwargs = cls.import_from_string(engine_kwargs_path)
else:
return {}
if not isinstance(engine_kwargs, dict):
- msg = f'`sqlalchemy_engine_kwargs` must be a dict, got {type(engine_kwargs).__name__!r}'
+ msg = f'`engine_kwargs` must be a dict, got {type(engine_kwargs).__name__!r}'
raise ConfigValidationError(msg)
return engine_kwargs
@classmethod
def _parse_engine(
cls,
- config: pytest.Config | None,
- engine: sa.Engine | AsyncEngine | None,
+ engine: AnyEngine | None,
+ engine_path: str | None,
engine_url: str | None,
+ engine_url_from_ini: str | None,
engine_kwargs: dict[str, Any],
- session_maker: SM | None,
- ) -> sa.Engine | AsyncEngine:
- if engine is None:
- if config is not None and (engine_path := config.getini('sqlalchemy_engine')):
- engine = cls.import_from_string(engine_path)
-
- elif engine_url is not None:
- if cls._is_async(engine_url):
- engine = create_async_engine(engine_url, **engine_kwargs)
- else:
- engine = sa.create_engine(engine_url, **engine_kwargs)
-
- elif config is not None and (engine_url_str := config.getini('sqlalchemy_engine_url')):
- if cls._is_async(engine_url_str):
- engine = create_async_engine(engine_url_str, **engine_kwargs)
- else:
- engine = sa.create_engine(engine_url_str, **engine_kwargs)
-
- elif session_maker is not None and isinstance(session_maker.kw['bind'], (sa.Engine, AsyncEngine)):
- engine = session_maker.kw['bind']
+ session_maker: AnySessionMaker | None,
+ ) -> AnyEngine:
+ new_engine: sa.Engine | AsyncEngine | None = None
+
+ if engine is not None:
+ new_engine = engine
+
+ elif engine_path is not None:
+ new_engine = cls.import_from_string(engine_path)
+ elif engine_url is not None:
+ if cls._is_async(engine_url):
+ new_engine = create_async_engine(engine_url, **engine_kwargs)
else:
- msg = "Couldn't configure engine: missing required config: sqlalchemy_engine, sqlalchemy_engine_url or sqlalchemy_session_maker"
- raise ConfigValidationError(msg)
+ new_engine = sa.create_engine(engine_url, **engine_kwargs)
+
+ elif engine_url_from_ini is not None:
+ if cls._is_async(engine_url_from_ini):
+ new_engine = create_async_engine(engine_url_from_ini, **engine_kwargs)
+ else:
+ new_engine = sa.create_engine(engine_url_from_ini, **engine_kwargs)
+
+ elif session_maker is not None and isinstance(session_maker.kw['bind'], (sa.Engine, AsyncEngine)):
+ new_engine = session_maker.kw['bind']
+
+ else:
+ msg = "Couldn't configure engine: missing required config: engine, engine_url or session_maker"
+ raise ConfigValidationError(msg)
- if not isinstance(engine, (sa.Engine, AsyncEngine)):
- msg = f'`sqlalchemy_engine` must be an Engine or AsyncEngine, got {type(engine).__name__!r}'
+ if not isinstance(new_engine, AnyEngine):
+ msg = f'`engine` must be an Engine or AsyncEngine, got {type(new_engine).__name__!r}'
raise ConfigValidationError(msg)
- return engine
+ return new_engine
@classmethod
- def _parse_metadata(cls, config: pytest.Config | None, metadata: sa.MetaData | Sequence[sa.MetaData] | None) -> Sequence[sa.MetaData]:
+ def _parse_metadata(cls, metadata: sa.MetaData | Sequence[sa.MetaData] | None, metadata_path: str | None) -> Sequence[sa.MetaData]:
if metadata is None:
- if config is None:
- return get_alembic_target_metadata()
-
- metadata_path = config.getini('sqlalchemy_metadata')
- if not metadata_path:
- return get_alembic_target_metadata()
- metadata = cls.import_from_string(metadata_path)
+ if metadata_path is None:
+ metadata = get_alembic_target_metadata()
+ else:
+ metadata = cls.import_from_string(metadata_path)
metadata_list: Sequence[Any]
if not isinstance(metadata, Sequence):
@@ -167,26 +202,30 @@ def _parse_metadata(cls, config: pytest.Config | None, metadata: sa.MetaData | S
for one_metadata in metadata_list:
if not isinstance(one_metadata, sa.MetaData):
- msg = f'`sqlalchemy_metadata` must contain metadata instance(s), got {type(one_metadata).__name__!r}'
+ msg = f'`metadata` must contain metadata instance(s), got {type(one_metadata).__name__!r}'
raise ConfigValidationError(msg)
return metadata_list
@classmethod
- def _parse_engine_scope(cls, config: pytest.Config | None, engine_scope: Literal['session', 'function']) -> Literal['session', 'function']:
- if config is not None:
- engine_scope = config.getini('sqlalchemy_engine_scope') or engine_scope
-
- if engine_scope not in ('session', 'function'):
- msg = f'`sqlalchemy_engine_scope` must be one of "session" or "function", got {engine_scope!r}'
+ def _parse_scope(
+ cls,
+ scope: Literal['session', 'function'] | None,
+ scope_from_ini: str,
+ ) -> Literal['session', 'function']:
+ if scope is None:
+ assert cls._validate_scope(scope_from_ini)
+ scope = scope_from_ini
+ if scope not in ('session', 'function'):
+ msg = f'`scope` must be one of "session" or "function", got {scope!r}'
raise ConfigValidationError(msg)
- return engine_scope
+ return scope
@classmethod
- def _load_orm_models(cls, config: pytest.Config | None, orm_loader: Callable[[], Any] | None) -> None:
+ def _load_orm_models(cls, orm_loader: Callable[[], Any] | None, orm_loader_path: str | None) -> None:
if orm_loader is not None:
orm_loader()
- elif config is not None and (orm_loader_path := config.getini('sqlalchemy_orm_loader')):
+ elif orm_loader_path is not None:
orm_loader = cls.import_from_string(orm_loader_path)
if callable(orm_loader):
orm_loader()
@@ -194,3 +233,60 @@ def _load_orm_models(cls, config: pytest.Config | None, orm_loader: Callable[[],
@classmethod
def _is_async(cls, url: str) -> bool:
return sa.make_url(url).get_dialect().is_async
+
+ @classmethod
+ def _validate_scope(cls, scope: str) -> TypeGuard[Literal['session', 'function']]:
+ return scope in ('session', 'function')
+
+
+@dataclass(slots=True)
+class IniEntry(TypeCheckDataclass):
+ session_maker: str | None = None
+ engine: str | None = None
+ engine_url: str | None = None
+ engine_kwargs: str | None = None
+ orm_loader: str | None = None
+ metadata: str | None = None
+ scope: str = 'session'
+ skip_db_management: bool = False
+
+ def __post_init__(self) -> None:
+ if self.scope not in ('session', 'function'):
+ msg = f'`scope` must be one of "session" or "function", got {self.scope!r}'
+ raise TypeError(msg)
+
+ @classmethod
+ def parse(cls, entry: Any) -> Self | None:
+ if isinstance(entry, dict):
+ try:
+ return cls(**entry)
+ except TypeError as error:
+ msg = f'Invalid configuration entry format: {entry}. Details: {error}'
+ raise pytest.UsageError(msg) from error
+
+ elif isinstance(entry, str):
+ entry = entry.lstrip()
+ assert isinstance(entry, str)
+
+ if not entry or entry.startswith('//'):
+ return None
+
+ try:
+ evaluated = json.loads(entry)
+ except ValueError as error:
+ msg = f"Could not parse configuration row: '{entry}'. Details: {error}"
+ raise pytest.UsageError(msg) from error
+
+ if not isinstance(evaluated, dict):
+ msg = f'Invalid configuration entry format: {entry}. Expected a dictionary, got {type(evaluated).__name__!r}.'
+ raise pytest.UsageError(msg)
+
+ try:
+ return cls(**evaluated)
+ except TypeError as error:
+ msg = f'Invalid configuration entry format: {entry}. Details: {error}'
+ raise pytest.UsageError(msg) from error
+
+ else:
+ msg = f'Invalid configuration entry type: {type(entry).__name__}. Expected dict or str.'
+ raise pytest.UsageError(msg)
diff --git a/src/pytest_sqlalchemy_alembic/plugin.py b/src/pytest_sqlalchemy_alembic/plugin.py
index d75d2ba..0df3fc6 100644
--- a/src/pytest_sqlalchemy_alembic/plugin.py
+++ b/src/pytest_sqlalchemy_alembic/plugin.py
@@ -1,7 +1,7 @@
from __future__ import annotations
import contextlib
-from collections.abc import AsyncGenerator, Generator
+from collections.abc import AsyncGenerator, Generator, Sequence
import pytest
from sqlalchemy import Engine
@@ -27,23 +27,17 @@ def pytest_addoption(parser: pytest.Parser) -> None:
'--nomigrations',
'--no-migrations',
action='store_true',
- dest='nomigrations',
+ dest='no_migrations',
default=False,
help='Disable alembic migrations on test setup',
)
- parser.addini('sqlalchemy_session_maker', 'Import path to a sessionmaker instance', 'string', default=None)
- parser.addini('sqlalchemy_engine', 'Import path to SQLAlchemy engine instance', 'string', default=None)
- parser.addini('sqlalchemy_engine_url', 'SQLAlchemy engine URL', 'string', default=None)
- parser.addini('sqlalchemy_engine_kwargs', 'Import path to a dict containing engine kwargs', 'string', default=None)
- parser.addini('sqlalchemy_orm_loader', 'Import path to module or callable that loads all ORM necessary models', 'string', default=None)
- parser.addini('sqlalchemy_metadata', 'Import path to SQLAlchemy metadata. Usually `metadata` attribute of a declarative base class', 'string', default=None)
- parser.addini('sqlalchemy_engine_scope', 'Defines at which scope test engine should be activated ("session" or "function")', 'string', default='session')
+ parser.addini('sqlalchemy_alembic_configs', '', 'linelist', default=[])
@pytest.fixture(scope='session')
-def sqlalchemy_alembic_plugin_config(pytestconfig: pytest.Config) -> PluginConfig:
+def sqlalchemy_alembic_plugin_configs(pytestconfig: pytest.Config) -> list[PluginConfig]:
"""Extension point to override config values for pytest-sqlalchemy-alembic plugin from python context."""
- return PluginConfig.build(pytestconfig)
+ return PluginConfig.build_list(pytestconfig)
def pytest_sessionstart(session: pytest.Session) -> None:
@@ -63,10 +57,17 @@ def pytest_sessionstart(session: pytest.Session) -> None:
@contextlib.contextmanager
-def _test_engine_ctx(cfg: PluginConfig, worker_id: str) -> Generator[BaseDatabaseManager, None, None]:
+def _test_engine_ctx(cfg: PluginConfig, worker_id: str, *, db_management: bool) -> Generator[BaseDatabaseManager, None, None]:
database_manager = cfg.database_manager(cfg.engine)
test_engine = database_manager.create_test_engine(worker_id, cfg.engine_kwargs)
+ if db_management:
+ database_manager.prepare_test_databases(create_db=cfg.create_db)
+ if cfg.no_migrations:
+ database_manager.create_tables(cfg.metadata)
+ else:
+ alembic_upgrade(database_manager.test_engine.url)
+
if cfg.session_maker is not None:
cfg.session_maker.configure(bind=test_engine)
@@ -79,10 +80,17 @@ def _test_engine_ctx(cfg: PluginConfig, worker_id: str) -> Generator[BaseDatabas
@contextlib.asynccontextmanager
-async def _async_test_engine_ctx(cfg: PluginConfig, worker_id: str) -> AsyncGenerator[BaseDatabaseManager]:
+async def _async_test_engine_ctx(cfg: PluginConfig, worker_id: str, *, db_management: bool) -> AsyncGenerator[BaseDatabaseManager]:
database_manager = cfg.database_manager(cfg.engine)
test_engine = database_manager.create_test_engine(worker_id, cfg.engine_kwargs)
+ if db_management:
+ await database_manager.async_prepare_test_databases(create_db=cfg.create_db)
+ if cfg.no_migrations:
+ await database_manager.async_create_tables(cfg.metadata)
+ else:
+ alembic_upgrade(database_manager.test_engine.url)
+
if cfg.session_maker is not None:
cfg.session_maker.configure(bind=test_engine)
@@ -96,69 +104,87 @@ async def _async_test_engine_ctx(cfg: PluginConfig, worker_id: str) -> AsyncGene
def setup_session_sync(
pytestconfig: pytest.Config,
- sqlalchemy_alembic_plugin_config: PluginConfig,
-) -> Generator[Engine | AsyncEngine, None, None]:
- """Session-scoped fixture to set up test database. Returns test engine instance."""
- cfg = sqlalchemy_alembic_plugin_config
+ sqlalchemy_alembic_plugin_configs: list[PluginConfig],
+) -> Generator[Sequence[Engine | AsyncEngine]]:
+ """Session-scoped fixture to set up test database. Returns a sequence of test engine instances."""
worker_id = resolve_worker_id(pytestconfig)
- with _test_engine_ctx(cfg, worker_id) as database_manager:
- database_manager.prepare_test_databases(create_db=pytestconfig.getoption('create_db'))
+ with contextlib.ExitStack() as stack:
+ test_engines = []
- if pytestconfig.getoption('nomigrations'):
- database_manager.create_tables(cfg.metadata)
- else:
- alembic_upgrade(database_manager.test_engine.url)
+ for cfg in sqlalchemy_alembic_plugin_configs:
+ database_manager = stack.enter_context(
+ _test_engine_ctx(cfg, worker_id, db_management=not cfg.skip_db_management),
+ )
+
+ test_engines.append(database_manager.test_engine)
- yield database_manager.test_engine
+ yield tuple(test_engines)
async def setup_session_async(
pytestconfig: pytest.Config,
- sqlalchemy_alembic_plugin_config: PluginConfig,
-) -> AsyncGenerator[Engine | AsyncEngine, None]:
- """Session-scoped fixture to set up test database. Returns test engine instance."""
- cfg = sqlalchemy_alembic_plugin_config
+ sqlalchemy_alembic_plugin_configs: list[PluginConfig],
+) -> AsyncGenerator[Sequence[Engine | AsyncEngine], None]:
+ """Session-scoped fixture to set up test database. Returns a sequence of test engine instances."""
worker_id = resolve_worker_id(pytestconfig)
- async with _async_test_engine_ctx(cfg, worker_id) as database_manager:
- await database_manager.async_prepare_test_databases(create_db=pytestconfig.getoption('create_db'))
+ async with contextlib.AsyncExitStack() as stack:
+ test_engines = []
- if pytestconfig.getoption('nomigrations'):
- await database_manager.async_create_tables(cfg.metadata)
- else:
- alembic_upgrade(database_manager.test_engine.url)
+ for cfg in sqlalchemy_alembic_plugin_configs:
+ database_manager = await stack.enter_async_context(
+ _async_test_engine_ctx(cfg, worker_id, db_management=not cfg.skip_db_management),
+ )
- yield database_manager.test_engine
+ test_engines.append(database_manager.test_engine)
+
+ yield tuple(test_engines)
def setup_function_sync(
pytestconfig: pytest.Config,
- sqlalchemy_alembic_plugin_config: PluginConfig,
-) -> Generator[Engine | AsyncEngine | None, None, None]:
- """Function-scoped fixture to set up test database. Returns test engine instance."""
- cfg = sqlalchemy_alembic_plugin_config
- if cfg.engine_scope != 'function':
- yield None
- return
-
+ sqlalchemy_alembic_plugin_configs: list[PluginConfig],
+) -> Generator[Sequence[Engine | AsyncEngine | None], None, None]:
+ """Function-scoped fixture to set up test database. Returns a sequence of test engine instances."""
worker_id = resolve_worker_id(pytestconfig)
- with _test_engine_ctx(cfg, worker_id) as database_manager:
- yield database_manager.test_engine
+ with contextlib.ExitStack() as stack:
+ test_engines: list[Engine | AsyncEngine | None] = []
+
+ for cfg in sqlalchemy_alembic_plugin_configs:
+ if cfg.scope != 'function':
+ test_engines.append(None)
+ continue
+
+ database_manager = stack.enter_context(
+ _test_engine_ctx(cfg, worker_id, db_management=False),
+ )
+
+ test_engines.append(database_manager.test_engine)
+
+ yield tuple(test_engines)
async def setup_function_async(
pytestconfig: pytest.Config,
- sqlalchemy_alembic_plugin_config: PluginConfig,
-) -> AsyncGenerator[Engine | AsyncEngine | None, None]:
- """Function-scoped fixture to set up test database. Returns test engine instance."""
- cfg = sqlalchemy_alembic_plugin_config
- if cfg.engine_scope != 'function':
- yield None
- return
-
+ sqlalchemy_alembic_plugin_configs: list[PluginConfig],
+) -> AsyncGenerator[Sequence[Engine | AsyncEngine | None], None]:
+ """Function-scoped fixture to set up test database. Returns a sequence of test engine instances."""
worker_id = resolve_worker_id(pytestconfig)
- async with _async_test_engine_ctx(cfg, worker_id) as database_manager:
- yield database_manager.test_engine
+ async with contextlib.AsyncExitStack() as stack:
+ test_engines: list[Engine | AsyncEngine | None] = []
+
+ for cfg in sqlalchemy_alembic_plugin_configs:
+ if cfg.scope != 'function':
+ test_engines.append(None)
+ continue
+
+ database_manager = await stack.enter_async_context(
+ _async_test_engine_ctx(cfg, worker_id, db_management=False),
+ )
+
+ test_engines.append(database_manager.test_engine)
+
+ yield tuple(test_engines)
diff --git a/src/pytest_sqlalchemy_alembic/tests/test_plugin.py b/src/pytest_sqlalchemy_alembic/tests/test_plugin.py
index a641a9c..59e4685 100644
--- a/src/pytest_sqlalchemy_alembic/tests/test_plugin.py
+++ b/src/pytest_sqlalchemy_alembic/tests/test_plugin.py
@@ -21,7 +21,8 @@ def test_db():
pytester.makeini(
f"""
[pytest]
- sqlalchemy_session_maker = {module_name}:SessionLocal
+ sqlalchemy_alembic_configs =
+ {{"session_maker": "{module_name}:SessionLocal"}}
"""
)
@@ -47,7 +48,8 @@ def test_db():
pytester.makeini(
f"""
[pytest]
- sqlalchemy_engine = {module_name}:engine
+ sqlalchemy_alembic_configs =
+ {{"engine": "{module_name}:engine"}}
"""
)
@@ -56,7 +58,7 @@ def test_db():
result.assert_outcomes(errors=0, passed=1)
-def test_plugin_load_reports_missing_required_sqlalchemy_config(pytester: pytest.Pytester) -> None:
+def test_plugin_load_empty_config_ok(pytester: pytest.Pytester) -> None:
pytester.makepyfile(
"""
def test_unconfigured():
@@ -66,5 +68,25 @@ def test_unconfigured():
result = pytester.runpytest_subprocess()
+ result.assert_outcomes(passed=1)
+
+
+def test_plugin_load_reports_missing_required_engine_config(pytester: pytest.Pytester) -> None:
+ pytester.makepyfile(
+ """
+ def test_under_configured():
+ pass
+ """
+ )
+ pytester.makeini(
+ """
+ [pytest]
+ sqlalchemy_alembic_configs =
+ {"scope": "session"}
+ """
+ )
+
+ result = pytester.runpytest_subprocess()
+
result.assert_outcomes(errors=1)
result.stdout.fnmatch_lines(["*Couldn't configure engine*"])
diff --git a/src/pytest_sqlalchemy_alembic/utils.py b/src/pytest_sqlalchemy_alembic/utils.py
index fef0f29..e4e4db2 100644
--- a/src/pytest_sqlalchemy_alembic/utils.py
+++ b/src/pytest_sqlalchemy_alembic/utils.py
@@ -1,8 +1,9 @@
import logging
import threading
from collections.abc import Sequence
+from dataclasses import dataclass, fields
from importlib import import_module
-from typing import Any
+from typing import Any, get_type_hints
import pytest
import sqlalchemy as sa
@@ -87,3 +88,19 @@ def import_string(import_path: str) -> Any:
except (ImportError, AttributeError) as e:
msg = f'Error importing {import_path!r}: {e}'
raise ImportError(msg) from e
+
+
+@dataclass
+class TypeCheckDataclass:
+ def __post_init__(self) -> None:
+ resolved_types = get_type_hints(self.__class__)
+
+ for field in fields(self):
+ if not field.init:
+ continue
+
+ value = getattr(self, field.name)
+ expected_type = resolved_types[field.name]
+ if not isinstance(value, expected_type):
+ msg = f"Field '{field.name}' must be of type {expected_type.__name__}, got {type(value).__name__} instead."
+ raise TypeError(msg)