Trigger events before and after drop table statements

[adrien-berchet]

Description

Add before_drop and after_drop events in alembic.ddl.impl.DefaultImpl.drop_table().

Fixes #1037

[CaselIT]

Hi

As mentioned in the template and in the code of conduct, we appreciate if before opening a PR an issue explaining the issue or use case was opened

[adrien-berchet]

Hi

As mentioned in the template and in the code of conduct, we appreciate if before opening a PR an issue explaining the issue or use case was opened

Hi @CaselIT
Sorry, I just submitted the related issue.

[adrien-berchet]

Note that this PR does not solve geoalchemy/geoalchemy2#374 because the table given as parameter to op.drop_table() is not the one received by the event, which has no attached column. This is because the drop_table operation is called with operation.to_table(operations.migration_context) as parameter instead of the actual table. I don't know why this behavior was chosen? I guess it was to support tables given as string names?

[zzzeek]

Note that this PR does not solve geoalchemy/geoalchemy2#374 because the table given as parameter to op.drop_table() is not the one received by the event, which has no attached column. This is because the drop_table operation is called with operation.to_table(operations.migration_context) as parameter instead of the actual table. I don't know why this behavior was chosen? I guess it was to support tables given as string names?

the SQLAlchemy Table object is not part of the op.create_table() or op.drop_table() API; these methods are passed string table names and an optional set of columns. so in that regard, the "events" don't make much sense, and the reason create_table() has support for the on-table-create events is to support triggering DDL for the Enum and Boolean datatypes when they make use of CHECK constraints or PG native ENUM types.

if you need full blown Table objects with state on them passed to the event then I'd not use op.drop_table(), I'd run my_table.drop(op.get_bind()), which will run the event handlers for that Table also.

[adrien-berchet]

Ok, I see. So I guess the best would be to define helpers in geoalchemy2 so that Alembic auto-generates migration scripts with my_table.drop(op.get_bind()) statements instead of op.drop_table("my_table_name") statements, right?

[zzzeek]

Ok, I see. So I guess the best would be to define helpers in geoalchemy2 so that Alembic auto-generates migration scripts with my_table.drop(op.get_bind()) statements instead of op.drop_table("my_table_name") statements, right?

does GeoAlchemy need to DROP other constructs that go along with the Table? the "Alembic" way to do this would be to have those DROP commands right in the migration script in a literal way. We have similar problems with PG ENUM, right now people add separate CREATE/DROP for PG ENUM objects and our autogenerate solution (someday) would be to render these statements in the migration script directly.

The autogenerate process can be customized so if you wanted to auto-generate additional geoalchemy-specific directives, there are ways to make that happen, https://alembic.sqlalchemy.org/en/latest/api/autogenerate.html#customizing-revision-generation contains some background on this.

The approach of having the directives right in the migration file makes it clear what's actually going on.

[zzzeek]

adding, my suggestion of mytable.drop(op.get_bind()) is not what I'd want in a finished product, that would be a hack to make something happen right now.

[adrien-berchet]

Ok I see, thanks for these details.
The table management in GeoAlchemy2 a bit messy because each dialect handle spatial types very differently.
For example, with SQLite the best would be to use func.DropTable('my_table') instead my_table.drop(). But I don't think we can intercept the drop to replace it by a DropSpatialTable statement (inside an event or whatever), is it? We considered it was not possible so we use the events to manually create/clean what we need (especially to handle the spatial indexes which use separate tables in SQLite). So it would be possible to add these statements in the migration scripts for SQLite but then the migration scripts would be dialect dependent, which is probably not what we want. To avoid that I am considering creating new specific operations, let's say op.DropSpatialTable(), as I already did for the columns with geoalchemy2.alembic_helpers.DropGeospatialColumn. And then the users will be able to add them to the autogenerate process as you mentioned. It's just a bit disappointing that we have to define the logic in two places (one for regular use and one specific for alembic) and I was hoping that triggering drop events would solve this.

[zzzeek]

yes we have this problem with ENUM and despite many years of people complaining about it, I haven't taken the time to figure out a single solution that can be the official "way to do it".

as far as intercepting "DROP TABLE" to write the SQL for that single statement in a different way, that in itself is straightforward and you wouldn't use events for that, you'd use the @compiles hook. It's not clear if what you need here is a different syntax for a single "DROP TABLE" SQL statement or if you have a series of individual directives that need to be emitted separately and to the degree they depend upon other structures inside of the table, so if I could see what SQL you want to emit for a few backends I might be able to recommend the best approach.

[adrien-berchet]

Hi @zzzeek

I tried to gather everything we need from the different docs, I hope it's accurate enough.
Thanks for your help!

PostgreSQL + PostGIS 1 (we should probably deprecate support of PostGIS < 2)

Create table

CREATE TABLE my_table (
	id SERIAL NOT NULL,
	PRIMARY KEY (id)
);
SELECT AddGeometryColumn('my_table_schema', 'my_table', 'geom', 4326, 'POINT', 2);

If spatial_index=True:

CREATE INDEX idx_my_table_geom ON my_table USING gist (geom);

Drop table

SELECT DropGeometryColumn('my_table_schema', 'my_table', 'geom');
DROP TABLE my_table;

or

SELECT DropGeometryTable('my_table_schema', 'my_table');

PostgreSQL + PostGIS 2 or PostGIS 3

Create table

CREATE TABLE my_table (
	id SERIAL NOT NULL, 
	geom geometry(POINT, 4326),
	PRIMARY KEY (id)
);

If spatial_index=True:

CREATE INDEX idx_my_table_geom ON my_table USING gist (geom);

Drop table

DROP TABLE my_table;

SQLite + SpatiaLite 4

Create table

CREATE TABLE my_table (
	id SERIAL NOT NULL,
	geom GEOMETRY,
	PRIMARY KEY (id)
);
SELECT RecoverGeometryColumn('my_table', 'geom', 4326, 'XY');

Note: Creating a dummy column then calling RecoverGeometryColumn is needed when dealing with complex constraints, see the test case geoalchemy2/tests/test_functional_sqlite.py::TestConstraint::test_insert.

If spatial_index=True:

SELECT CreateSpatialIndex('my_table', 'geom');

Drop table

SELECT DropGeoTable('my_table');

SQLite + SpatiaLite 5

Create table

CREATE TABLE my_table (
	id SERIAL NOT NULL, 
	geom GEOMETRY,
	PRIMARY KEY (id)
);
SELECT RecoverGeometryColumn('my_table', 'geom', 4326, 'XY');

Note: Creating a dummy column then calling RecoverGeometryColumn is needed when dealing with complex constraints, see the test case geoalchemy2/tests/test_functional_sqlite.py::TestConstraint::test_insert.

If spatial_index=True:

SELECT CreateSpatialIndex('my_table', 'geom');

Drop table

SELECT DropTable(NULL, 'my_table');

Note: This will automatically remove the table and all associated indexes (which are tables themselves) but it is only available for Spatialite >= 5.

Rename table

SELECT RenameTable(NULL, 'my_table', 'my_new_table');

Rename column

SELECT RenameColumn(NULL, 'my_table', 'old_col_name', 'new_col_name');

Miscellaneous

• Need SQLite >= 3.25 (2018-09-15) to be able to rename tables and columns.

MySQL (not supported yet)

Create table

CREATE TABLE my_table (
	id SERIAL NOT NULL,
	geom GEOMETRY NOT NULL SRID 4326
);

Note that the geometry type is not given here.

If spatial_index=True, several queries are possible:

CREATE TABLE my_table (
	id SERIAL NOT NULL,
	geom GEOMETRY NOT NULL SRID 4326,
	SPATIAL INDEX(geom)
);

or

CREATE TABLE my_table (
	id SERIAL NOT NULL,
	geom GEOMETRY NOT NULL SRID 4326
);
ALTER TABLE my_table ADD SPATIAL INDEX(geom);

or

CREATE TABLE geom (
	id SERIAL NOT NULL,
	geom GEOMETRY NOT NULL SRID 4326
);
CREATE SPATIAL INDEX geom ON my_table(geom);

Drop table

DROP TABLE my_table;

[zzzeek]

So it looks like maybe you would be compiling on DropTable, but from an Alembic point of view, the directive doesn't give you the information you need to run these commands:

op.drop_table("my_table")

that directive simply does not supply the information you need to know that there are geometry constructs on the table.

I think the cleanest approach for geoalchemy would be that you have your own directives, it seems like for the CREATE side things are column-specific, while on the DROP side they are table specific:

op.create_geometry_column("my_table",  Column("gist", GEOMETRY), ...)

op.drop_geometry_table("my_table", extra_geom_info = ...)

With this kind of API, you can add any kinds of arguments and options you need explicitly, and you then have a straightforward way to run all the DDL you need.

A complete example of how to create a custom op directive is at: https://alembic.sqlalchemy.org/en/latest/api/operations.html#operation-plugins . Decorators like @Operations.register_operation() and @Operations.implementation_for() allow you to construct the entire directive and how it should proceed on a Connection directly.

I am getting the impression that for the moment, you have "CREATE" working because you get the whole Table and you can look around for geometry information. So you can start with just the drop operation above.

For autogenerate support, there are two APIs that may be of use. The first is to be able to support detection of a totally new kind of object, like a trigger or a sequence, that's documented at https://alembic.sqlalchemy.org/en/latest/api/autogenerate.html#autogenerating-custom-operation-directives . However, at least for drop_table(), there is already going to be a drop_table directive spit out when a table is removed from the model, so you want to intercept that and rewrite it with your op.drop_geometry_table() directive. You can do this using the "rewriter" system which is documented at https://alembic.sqlalchemy.org/en/latest/api/autogenerate.html#fine-grained-autogenerate-generation-with-rewriters . To paraphrase the example that's there:

from alembic.autogenerate import rewriter
from alembic.operations import ops

writer = rewriter.Rewriter()

@writer.rewrites(ops.DropTableOp)
def add_column(context, revision, op):
    if has_geometry_things(op.table):
        return [
              DropGeometryTableOp(op.table_name, extra_geometry_things)
       ]
    else:
        return op

your DropGeometryTableOp will know how to write its function in a migration script using @renderers.dispatch_for, documented at https://alembic.sqlalchemy.org/en/latest/api/autogenerate.html#creating-a-render-function .

this is a lot to take in but alembic should have everything you need to make a first class migration API here. There is already one project using these APIs extensively which you can see at https://olirice.github.io/alembic_utils/. This user came to me a few years ago and I got them started with all these APIs and they seem to have all worked out.

[adrien-berchet]

Ok, I think I see what to do. I just have two questions:

• what should I expose to the user? The writer object? So the user would just have to write this?

  from geoalchemy2.alembic_helpers import writer
  
  def run_migrations_online():
      # ...
      with connectable.connect() as connection:
          context.configure(
              connection=connection,
              target_metadata=target_metadata,
              process_revision_directives=writer,
          )
  
          with context.begin_transaction():
              context.run_migrations()

• how can I test this? I tried with something like this but the writer is ignored:

  def test_migration(conn, metadata):
      alembic_table = Table(
          # ...
      )
  
      mc = MigrationContext.configure(
          conn,
          opts={
              "process_revision_directives": alembic_helpers.writer,
          },
      )
      migration_script = produce_migrations(mc, metadata)
      op = Operations(mc)
  
      # Autogenerate the migration script as a string and execute it
      eval(compile(upgrade_script.replace("    ", ""), "upgrade_script.py", "exec"))

Thanks!

[zzzeek]

Ok, I think I see what to do. I just have two questions:

* what should I expose to the user? The `writer` object? So the user would just have to write this?
  ```python
  from geoalchemy2.alembic_helpers import writer
  
  def run_migrations_online():
      # ...
      with connectable.connect() as connection:
          context.configure(
              connection=connection,
              target_metadata=target_metadata,
              process_revision_directives=writer,
          )
  
          with context.begin_transaction():
              context.run_migrations()
  ```

I think for the moment that's how it would have to be if rewriter is what works best here.

* how can I test this? I tried with something like this but the writer is ignored:
  ```python
  def test_migration(conn, metadata):
      alembic_table = Table(
          # ...
      )
  
      mc = MigrationContext.configure(
          conn,
          opts={
              "process_revision_directives": alembic_helpers.writer,
          },
      )
      migration_script = produce_migrations(mc, metadata)
      op = Operations(mc)
  
      # Autogenerate the migration script as a string and execute it
      eval(compile(upgrade_script.replace("    ", ""), "upgrade_script.py", "exec"))
  ```

produce_migrations gives you the MigrationScript structure alone, that's before a hook like this would be called. While you can call the rewriter directly on that, if you wanted to test the integration into "process_revision_directives" you'd want to use the RevisionContext object.

Clearly you can test at higher or lower levels here but the most end-to-end test of all the APIs would be if you were calling into command.revision(). It returns the Script structure that references the generated module.

Thanks!

[adrien-berchet]

Ok, I think I am almost good now, thanks to your help.

As far as I can see, I just have a last issue with spatial indexes.
In geoalchemy/geoalchemy2#374 , I added the writer and I use it in the test https://github.com/adrien-berchet/geoalchemy2/blob/fix_sqlite_reflection/tests/test_alembic_migrations.py#L301 using command.revision. This works well for the PostgreSQL dialect but not with the SQLite one when renaming spatial columns. In this case, the migration script is missing operations to create/drop a spatial index.
Here are the migration scripts produced for these 2 dialects:

Postgresql

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_geospatial_column('new_spatial_table', sa.Column('new_geom_with_idx', Geometry(geometry_type='LINESTRING', srid=4326, spatial_index=False, from_text='ST_GeomFromEWKT', name='geometry'), nullable=True))
    op.add_geospatial_column('new_spatial_table', sa.Column('new_geom_without_idx', Geometry(geometry_type='LINESTRING', srid=4326, spatial_index=False, from_text='ST_GeomFromEWKT', name='geometry'), nullable=True))
    op.drop_geospatial_index('idx_new_spatial_table_geom_with_idx', table_name='new_spatial_table', column_name='geom_with_idx')
    op.create_geospatial_index('idx_new_spatial_table_new_geom_with_idx', 'new_spatial_table', ['new_geom_with_idx'], unique=False, postgresql_using='gist', postgresql_ops={})
    op.drop_geospatial_column('new_spatial_table', 'geom_without_idx')
    op.drop_geospatial_column('new_spatial_table', 'geom_with_idx')
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_geospatial_column('new_spatial_table', sa.Column('geom_with_idx', Geometry(geometry_type='LINESTRING', srid=4326, spatial_index=False, from_text='ST_GeomFromEWKT', name='geometry'), autoincrement=False, nullable=True))
    op.add_geospatial_column('new_spatial_table', sa.Column('geom_without_idx', Geometry(geometry_type='LINESTRING', srid=4326, spatial_index=False, from_text='ST_GeomFromEWKT', name='geometry'), autoincrement=False, nullable=True))
    op.drop_geospatial_index('idx_new_spatial_table_new_geom_with_idx', table_name='new_spatial_table', postgresql_using='gist', postgresql_ops={}, column_name='new_geom_with_idx')
    op.create_geospatial_index('idx_new_spatial_table_geom_with_idx', 'new_spatial_table', ['geom_with_idx'], unique=False, postgresql_using='gist', postgresql_ops={})
    op.drop_geospatial_column('new_spatial_table', 'new_geom_without_idx')
    op.drop_geospatial_column('new_spatial_table', 'new_geom_with_idx')
    # ### end Alembic commands ###

SQLite

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_geospatial_column('new_spatial_table', sa.Column('new_geom_with_idx', Geometry(geometry_type='LINESTRING', srid=4326, spatial_index=False, from_text='ST_GeomFromEWKT', name='geometry'), nullable=True))
    op.add_geospatial_column('new_spatial_table', sa.Column('new_geom_without_idx', Geometry(geometry_type='LINESTRING', srid=4326, spatial_index=False, from_text='ST_GeomFromEWKT', name='geometry'), nullable=True))
    op.create_geospatial_index('idx_new_spatial_table_new_geom_with_idx', 'new_spatial_table', ['new_geom_with_idx'], unique=False, postgresql_using='gist', postgresql_ops={})
    op.drop_geospatial_column('new_spatial_table', 'geom_without_idx')
    op.drop_geospatial_column('new_spatial_table', 'geom_with_idx')
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_geospatial_column('new_spatial_table', sa.Column('geom_with_idx', Geometry(geometry_type='LINESTRING', srid=4326, spatial_index=False, from_text='ST_GeomFromEWKT', name='geometry'), nullable=True))
    op.add_geospatial_column('new_spatial_table', sa.Column('geom_without_idx', Geometry(geometry_type='LINESTRING', srid=4326, spatial_index=False, from_text='ST_GeomFromEWKT', name='geometry'), nullable=True))
    op.drop_geospatial_index('idx_new_spatial_table_new_geom_with_idx', table_name='new_spatial_table', postgresql_using='gist', postgresql_ops={}, column_name='new_geom_with_idx')
    op.drop_geospatial_column('new_spatial_table', 'new_geom_without_idx')
    op.drop_geospatial_column('new_spatial_table', 'new_geom_with_idx')
    # ### end Alembic commands ###

As far as I can see, it is because the batch operation renames the temporary table using a regular ALTER TABLE ... RENAME TO ..., which does not renames the related indexes (the SpatiaLite function SELECT RenameTable() should be called to fix this). After that, the table is not reflected properly to retrieve the indexes (since the indexes are still associated to the temporary table, which was actually deleted), and thus the indexes are not detected and added to the migration script. Unfortunately, https://github.com/sqlalchemy/alembic/blob/main/alembic/operations/batch.py#L457 directly calls op_impl.rename_table, which it seems I can't change, can I? I could monkey patch alembic.ddl.SQLiteImpl to add a SQLite-specific rename_table() method but I would like to avoid that. Do you have an idea to fix this?

[zzzeek]

the code example you have there isn't using batch.....but overall yeah all this great extensibility is kind of non-existent for batch mode. if the issue is just the rename_table() operation, you could add a directive to override alembic.ddl.base.RenameTable:

@compiles(RenameTable, "sqlite")
def visit_rename_table(
    element: "RenameTable", compiler: "DDLCompiler", **kw
) -> str:
    # your code here
    return "%s RENAME TO %s" % (
        alter_table(compiler, element.table_name, element.schema),
        format_table_name(compiler, element.new_table_name, element.schema),
    )

but you just have a table name there. So maybe if you had some global registry of table names -> which of them are geo, something like that, to tell when to use the "rename geo" version of things. or maybe it works in all cases, dunno.

[adrien-berchet]

the code example you have there isn't using batch

Ups sorry, I didn't commit the last change. The batch operation is here: https://github.com/adrien-berchet/geoalchemy2/blob/fix_sqlite_reflection/geoalchemy2/alembic_helpers.py#L141

but overall yeah all this great extensibility is kind of non-existent for batch mode. if the issue is just the rename_table() operation, you could add a directive to override alembic.ddl.base.RenameTable

Ah yeah I could try that. I am not a fan of global registries but let's try :)

Thanks again!

[zzzeek]

why use batch mode internally like that? alembic's documented approach has the batch operation publicly inside the migration script itself. the reason is so that a series of changes to a single table can all be done at once, since batch involves copying the entire table. if someone has a SQLite database important enough that they actually need to run migrations on it, it might also have millions of rows of data. allowing the batch directive to be in the migration script also allows space for some additional directives like constraints and things that batch mode wont detect; there's probably geo-specific concepts that can benefit from this part of it being exposed.

[adrien-berchet]

I naively tried to use batch mode here because SQLite can not drop a column directly, but that's indeed not a satisfactory solution.
Anyway, I was able to call SELECT RenameTable() instead of ALTER TABLE ... RENAME TO ... and it works ALMOST properly with spatialite >= 5 (for older versions I would have to write the equivalent function but I think I will not do it and just wait for people to use newer version 😇 ). My last issue is still that the spatial indexes are not reflected by sqlalchemy for the SQLite dialect. So in alembic.autogenerate.compare._compare_indexes_and_uniques() the conn_indexes are empty and thus the migration scripts are missing the related statements. I can't see any way to customize index reflection in SQLAlchemy so I am not sure how to do this. Maybe I could just insert a create_index statement in migration scripts when there is a create_column statement and the added column has spatial_index=True? Is that possible?

[adrien-berchet]

Hi there,
I updated geoalchemy/geoalchemy2#374 and now the test passes with both postgresql and sqlite dialects, so I guess everything is fine now (the tests pass locally but the CI fails because it needs the new version of Alembic). I had to monkey patch the index reflection mechanism of SQLAlchemy to reflect spatial indexes with SQLite dialect properly. I think it's no big deal since it just adds a specific behavior for these spatial indexes. Maybe in the future it would be nice to add new events so we can deal with such cases in a better way.
Anyway, as far as I am concerned, I don't need anything else from the Alembic side so it can be released.
Thank you very much for your help!

[zzzeek]

hey @adrien-berchet -

sorry I didnt reply previously. i have no time to do anything these days and it looked like you were on the right track. looks like 1.7.8 has a lot to release.

that said, we should probably still merge the actual PR here since it's already done and nice to have.

[sqla-tester]

OK, this is sqla-tester setting up my work on behalf of zzzeek to try to get revision ea44e7f of this pull request into gerrit so we can run tests and reviews and stuff

[sqla-tester]

New Gerrit review created for change ea44e7f: https://gerrit.sqlalchemy.org/c/sqlalchemy/alembic/+/3882

[adrien-berchet]

sorry I didnt reply previously. i have no time to do anything these days and it looked like you were on the right track. looks like 1.7.8 has a lot to release.

No problem, you already helped me a lot and I was finally able to make it work :)
For the monkey patch part, should I open an issue on SQLAlchemy about adding a mechanism to be able to customize index reflection?

that said, we should probably still merge the actual PR here since it's already done and nice to have.

Ok, as you prefer. In that case I suggest to make it clear in the doc that the table passed to these events is not the actual table but a dummy one (without all the columns and indexes attached).

[CaselIT]

should I open an issue on SQLAlchemy about adding a mechanism to be able to customize index reflection?

yes please, please open an issue "use case" otherwise it will get lost :)

[zzzeek]

sorry I didnt reply previously. i have no time to do anything these days and it looked like you were on the right track. looks like 1.7.8 has a lot to release.

No problem, you already helped me a lot and I was finally able to make it work :) For the monkey patch part, should I open an issue on SQLAlchemy about adding a mechanism to be able to customize index reflection?

That's definitely not possible without writing a dialect subclass. Can you provide specifics on this index reflection? in alembic, this should be done using a custom comparison plugin, that would compare these additional indexes: https://alembic.sqlalchemy.org/en/latest/api/autogenerate.html#registering-a-comparison-function

that said, we should probably still merge the actual PR here since it's already done and nice to have.

Ok, as you prefer. In that case I suggest to make it clear in the doc that the table passed to these events is not the actual table but a dummy one (without all the columns and indexes attached).

good point

[CaselIT]

That's definitely not possible without writing a dialect subclass.

off the top of my head my idea was that we could accept a function that gets passed with the connection, the indexes loaded and the table(s) reflected, and could modify the loaded indexes? Or alternatively it could make sense have something like this in the dialect events (so for every reflection method?)

probably not something for v2, but maybe could stay there as an idea?

[zzzeek]

That's definitely not possible without writing a dialect subclass.

off the top of my head my idea was that we could accept a function that gets passed with the connection, the indexes loaded and the table(s) reflected, and could modify the loaded indexes? Or alternatively it could make sense have something like this in the dialect events (so for every reflection method?)

probably not something for v2, but maybe could stay there as an idea?

it depends on what the problem is. if index reflection is producing the wrong answer, we should just fix that. if it's omitting things because they rely on plugins, they should write custom comparators on the alembic side.

basically I dont think reflection should have injectable alt-code.

[adrien-berchet]

should I open an issue on SQLAlchemy about adding a mechanism to be able to customize index reflection?

yes please, please open an issue "use case" otherwise it will get lost :)

Ok, I will do that 👍

That's definitely not possible without writing a dialect subclass. Can you provide specifics on this index reflection? in alembic, this should be done using a custom comparison plugin, that would compare these additional indexes: https://alembic.sqlalchemy.org/en/latest/api/autogenerate.html#registering-a-comparison-function

The spatial indexes must be queried in a specific way with SQLite dialect (see https://github.com/adrien-berchet/geoalchemy2/blob/fix_sqlite_reflection/geoalchemy2/alembic_helpers.py#L57) and added to regular indexes. Nevertheless, these indexes should not be propagated during copies (that's why I use _column_flag = True) because these indexes need specific processes to be created with SQLite dialect (with postgresql they are just created with a CREATE INDEX ... statement).
I tried a bit with comparators but I was not able to make it work. But I will try to take some time to try again later.

off the top of my head my idea was that we could accept a function that gets passed with the connection, the indexes loaded and the table(s) reflected, and could modify the loaded indexes? Or alternatively it could make sense have something like this in the dialect events (so for every reflection method?)

This could be very convenient indeed. In both cases (function or events), I could just use the function I wrote for the monkey patch except it would not call the 'normal behavior' at the beginning since it would get the regular indexes as parameters.

[adrien-berchet]

should I open an issue on SQLAlchemy about adding a mechanism to be able to customize index reflection?

yes please, please open an issue "use case" otherwise it will get lost :)

sqlalchemy/sqlalchemy#8080

[sqla-tester]

Gerrit review https://gerrit.sqlalchemy.org/c/sqlalchemy/alembic/+/3882 has been merged. Congratulations! :)