diff options
author | CaselIT <cfederico87@gmail.com> | 2023-04-06 22:16:41 +0200 |
---|---|---|
committer | Federico Caselli <cfederico87@gmail.com> | 2023-04-10 21:58:17 +0200 |
commit | 157c521736f1c9cfceb9b3a6ecf17f782d358c46 (patch) | |
tree | 8c38a3be951bee504a51625dc0ca01d0fb888f4c /alembic/autogenerate | |
parent | 3d9b1128cd6bf03ecb45003587c0eedfb9552b07 (diff) | |
download | alembic-157c521736f1c9cfceb9b3a6ecf17f782d358c46.tar.gz |
Use column sort in index compare on postgresql
Added support for autogenerate comparison of indexes on PostgreSQL which
include SQL sort option, such as ``ASC`` or ``NULLS FIRST``.
Fixes: #1213
Change-Id: I3ddcb647928d948e41462b1c889b1cbb515ace4f
Diffstat (limited to 'alembic/autogenerate')
-rw-r--r-- | alembic/autogenerate/compare.py | 24 |
1 files changed, 23 insertions, 1 deletions
diff --git a/alembic/autogenerate/compare.py b/alembic/autogenerate/compare.py index 4f5126f..85cb426 100644 --- a/alembic/autogenerate/compare.py +++ b/alembic/autogenerate/compare.py @@ -8,6 +8,7 @@ from typing import cast from typing import Dict from typing import Iterator from typing import List +from typing import Mapping from typing import Optional from typing import Set from typing import Tuple @@ -19,6 +20,7 @@ from sqlalchemy import inspect from sqlalchemy import schema as sa_schema from sqlalchemy import text from sqlalchemy import types as sqltypes +from sqlalchemy.sql import expression from sqlalchemy.util import OrderedSet from alembic.ddl.base import _fk_spec @@ -278,15 +280,35 @@ def _compare_tables( upgrade_ops.ops.append(modify_table_ops) +_IndexColumnSortingOps: Mapping[str, Any] = util.immutabledict( + { + "asc": expression.asc, + "desc": expression.desc, + "nulls_first": expression.nullsfirst, + "nulls_last": expression.nullslast, + "nullsfirst": expression.nullsfirst, # 1_3 name + "nullslast": expression.nullslast, # 1_3 name + } +) + + def _make_index(params: Dict[str, Any], conn_table: Table) -> Optional[Index]: exprs: list[Union[Column[Any], TextClause]] = [] + sorting = params.get("column_sorting") + for num, col_name in enumerate(params["column_names"]): item: Union[Column[Any], TextClause] if col_name is None: assert "expressions" in params - item = text(params["expressions"][num]) + name = params["expressions"][num] + item = text(name) else: + name = col_name item = conn_table.c[col_name] + if sorting and name in sorting: + for operator in sorting[name]: + if operator in _IndexColumnSortingOps: + item = _IndexColumnSortingOps[operator](item) exprs.append(item) ix = sa_schema.Index( params["name"], *exprs, unique=params["unique"], _table=conn_table |