How to run migrations from code with new 1.16 version? #1674
-
Now official docs recommends to move a part of settings, including @pytest.fixture(scope="module")
def alembic_config() -> config.Config:
return config.Config("alembic.ini") But now I'm facing with error: I tried to change to: Should I copy |
Beta Was this translation helpful? Give feedback.
Replies: 2 comments 2 replies
-
That would be a bug, no existing API use should break. Can you provide complete stack trace at least, if not also code to reproduce the problem ? |
Beta Was this translation helpful? Give feedback.
-
alembic.ini:[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = ERROR
handlers = console
qualname =
[logger_sqlalchemy]
level = ERROR
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = ERROR
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S At pyproject.toml:[tool.alembic]
script_location = "%(here)s/alembic"
prepend_sys_path = [
"src"
]
[[tool.alembic.post_write_hooks]]
name = "ruff"
type = "exec"
executable = "%(here)s/.venv/bin/ruff"
options = "check --fix REVISION_SCRIPT_FILENAME" alembic/env.py:import asyncio
from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.ext.asyncio import async_engine_from_config
from adapters.database.engine import Base
from adapters.database.models import * # noqa: F403
from alembic import context
from config import settings
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
config.set_main_option("sqlalchemy.url", settings.dsn)
target_metadata = Base.metadata
def do_run_migrations(connection):
context.configure(connection=connection, target_metadata=target_metadata)
with context.begin_transaction():
context.run_migrations()
async def run_async_migrations():
connectable = async_engine_from_config(
config.get_section(config.config_ini_section),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
async with connectable.connect() as connection:
await connection.run_sync(do_run_migrations)
await connectable.dispose()
def run_migrations_online():
connectable = config.attributes.get("connection", None)
if connectable is None:
asyncio.run(run_async_migrations())
else:
do_run_migrations(connectable)
run_migrations_online() Pytest fixtures for migrations:@pytest.fixture(scope="module")
def alembic_config() -> config.Config:
return config.Config("alembic.ini")
def run_alembic_upgrade(connection: Connection, alembic_config: config.Config):
alembic_config.attributes["connection"] = connection
command.upgrade(alembic_config, "heads")
def run_alembic_downgrade(
connection: Connection, alembic_config: config.Config
):
alembic_config.attributes["connection"] = connection
command.downgrade(alembic_config, "base")
@pytest.fixture(scope="module")
async def run_migrations_example(
async_engine: AsyncEngine,
alembic_config: config.Config,
) -> AsyncGenerator[None]:
async with async_engine.begin() as conn:
await conn.run_sync(run_alembic_upgrade, alembic_config)
yield
async with async_engine.begin() as conn:
await conn.run_sync(run_alembic_downgrade, alembic_config)
# Usage in tests:
@pytest.mark.usefixtures("run_migrations_example")
async def test_migrations():
assert True Stack trace: @functools.wraps(fixture)
def _asyncgen_fixture_wrapper(request: FixtureRequest, **kwargs: Any):
func = _perhaps_rebind_fixture_func(fixture, request.instance)
event_loop_fixture_id = _get_event_loop_fixture_id_for_async_fixture(
request, func
)
event_loop = request.getfixturevalue(event_loop_fixture_id)
kwargs.pop(event_loop_fixture_id, None)
gen_obj = func(**_add_kwargs(func, kwargs, event_loop, request))
async def setup():
res = await gen_obj.__anext__() # type: ignore[union-attr]
return res
context = contextvars.copy_context()
setup_task = _create_task_in_context(event_loop, setup(), context)
> result = event_loop.run_until_complete(setup_task)
.venv/lib/python3.13/site-packages/pytest_asyncio/plugin.py:345:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/usr/local/lib/python3.13/asyncio/base_events.py:719: in run_until_complete
return future.result()
.venv/lib/python3.13/site-packages/pytest_asyncio/plugin.py:340: in setup
res = await gen_obj.__anext__() # type: ignore[union-attr]
tests/conftest.py:69: in app
await conn.run_sync(run_alembic_upgrade, alembic_config)
.venv/lib/python3.13/site-packages/sqlalchemy/ext/asyncio/engine.py:887: in run_sync
return await greenlet_spawn(
.venv/lib/python3.13/site-packages/sqlalchemy/util/_concurrency_py3k.py:190: in greenlet_spawn
result = context.switch(*args, **kwargs)
tests/conftest.py:51: in run_alembic_upgrade
command.upgrade(alembic_config, "heads")
.venv/lib/python3.13/site-packages/alembic/command.py:463: in upgrade
script = ScriptDirectory.from_config(config)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
cls = <class 'alembic.script.base.ScriptDirectory'>, config = <alembic.config.Config object at 0x7f50e5138830>
@classmethod
def from_config(cls, config: Config) -> ScriptDirectory:
"""Produce a new :class:`.ScriptDirectory` given a :class:`.Config`
instance.
The :class:`.Config` need only have the ``script_location`` key
present.
"""
script_location = config.get_alembic_option("script_location")
if script_location is None:
> raise util.CommandError(
"No 'script_location' key found in configuration."
)
E alembic.util.exc.CommandError: No 'script_location' key found in configuration.
.venv/lib/python3.13/site-packages/alembic/script/base.py:175: CommandError config.Config with pyproject.toml:@pytest.fixture(scope="module")
def alembic_config() -> config.Config:
return config.Config("pyproject.toml") Stack trace: /usr/local/lib/python3.13/asyncio/base_events.py:719: in run_until_complete
return future.result()
.venv/lib/python3.13/site-packages/pytest_asyncio/plugin.py:340: in setup
res = await gen_obj.__anext__() # type: ignore[union-attr]
tests/conftest.py:69: in app
await conn.run_sync(run_alembic_upgrade, alembic_config)
.venv/lib/python3.13/site-packages/sqlalchemy/ext/asyncio/engine.py:887: in run_sync
return await greenlet_spawn(
.venv/lib/python3.13/site-packages/sqlalchemy/util/_concurrency_py3k.py:190: in greenlet_spawn
result = context.switch(*args, **kwargs)
tests/conftest.py:51: in run_alembic_upgrade
command.upgrade(alembic_config, "heads")
.venv/lib/python3.13/site-packages/alembic/command.py:463: in upgrade
script = ScriptDirectory.from_config(config)
.venv/lib/python3.13/site-packages/alembic/script/base.py:173: in from_config
script_location = config.get_alembic_option("script_location")
.venv/lib/python3.13/site-packages/alembic/config.py:440: in get_alembic_option
if self.file_config.has_option(self.config_ini_section, name):
.venv/lib/python3.13/site-packages/sqlalchemy/util/langhelpers.py:1226: in __get__
obj.__dict__[self.__name__] = result = self.fget(obj)
.venv/lib/python3.13/site-packages/alembic/config.py:235: in file_config
compat.read_config_parser(file_config, [self._config_file_path])
.venv/lib/python3.13/site-packages/alembic/util/compat.py:144: in read_config_parser
return file_config.read(file_argument, encoding="locale")
/usr/local/lib/python3.13/configparser.py:735: in read
self._read(fp, filename)
/usr/local/lib/python3.13/configparser.py:1050: in _read
ParsingError._raise_all(self._read_inner(fp, fpname))
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
exceptions = <list_iterator object at 0x7fb7be920c10>
@staticmethod
def _raise_all(exceptions: Iterable['ParsingError']):
"""
Combine any number of ParsingErrors into one and raise it.
"""
exceptions = iter(exceptions)
with contextlib.suppress(StopIteration):
> raise next(exceptions).combine(exceptions)
E configparser.ParsingError: Source contains parsing errors: PosixPath('pyproject.toml')
E [line 19]: ']\n'
E [line 30]: ']\n'
E [line 38]: ']\n'
E [line 42]: ']\n'
E [line 47]: ']\n'
E [line 51]: ']\n'
E [line 59]: ']\n'
E [line 73]: ']\n'
E [line 78]: ']\n'
E [line 84]: ']\n'
/usr/local/lib/python3.13/configparser.py:334: ParsingError |
Beta Was this translation helpful? Give feedback.
OK can you clarify that your code before you tried to use pyproject.toml worked fine? that is, alembic 1.16 should introduce no regressions to existing code. that's my most immediate concern, that you upgraded to alembic and your code without any changes didn't break.
then, when you wanted to change to use pyproject.toml, that's fine, but the docs probably dont make clear that you need to send it as a separate keyword argument:
the keyword arg is there at https://alembic.sqlalchemy.org/en/latest/api/config.html#alembic.config.Config however isn't documented so that should be added. I think you do have to indicate it explicitly though. will upda…