我正在尝试sqlalchemy使用 That works 自动创建物化视图sqlalchemy_utils.view.create_materialized_view()，但我也希望它也自动为其创建索引。sqlalchemy_utils为此提供了一种机制，但需要一个sa.Index需要对象的Column对象。该Column对象是从一个可选对象中创建的，但模型上不存在该对象以在创建索引时将其传递给该对象。
class MVCustomerSearch(Base):
__table__ = create_materialized_view(
"mv_customer_search",
select(
[
TSubStation.ixSubStation.label("ixSubStation"),
TLocation.ixLocation.label("ixLocation"),
TCustomer.ixCustomer.label("ixCustomer"),
(
func.to_tsvector("english", func.coalesce(TLocation.sLocation, ""))
+ func.to_tsvector(
"english", func.coalesce(TCustomer.sCustomer, "")
)
+ func.to_tsvector(
"english", func.coalesce(TSubStation.sSubStation, "")
)
).label("tsv"),
],
).select_from(
join(
TCustomer, TLocation, TCustomer.ixCustomer == TLocation.ixCustomer
).join(TSubStation, TSubStation.ixLocation == TLocation.ixLocation)
),
metadata,
indexes=(
db.Index(
"idx_fts_customer_search",
# This needs to be a Column and won't work.
"MVCustomerSearch.tsv",
postgresql_using="gin",
)
),
)
customer = db.relationship(
"TCustomer",
uselist=False,
primaryjoin="TCustomer.ixCustomer==MVCustomerSearch.ixCustomer",
foreign_keys="TCustomer.ixCustomer",
)
location = db.relationship(
"TLocation",
uselist=False,
primaryjoin="TLocation.ixLocation==MVCustomerSearch.ixLocation",
foreign_keys="TLocation.ixLocation",
)
substation = db.relationship(
"TSubStation",
uselist=False,
primaryjoin="TSubStation.ixSubStation==MVCustomerSearch.ixSubStation",
foreign_keys="TSubStation.ixSubStation",
)
如果我在模型存在后模拟创建索引，它将起作用。
In [31]: from .models.dbviews import MVCustomerSearch
In [32]: idx = db.Index(
...: "idx_fts_customer_search",
...: MVCustomerSearch.tsv,
...: postgresql_using="gin",
...: )
In [33]: print(CreateIndex(idx).compile(dialect=postgresql.dialect()))
CREATE INDEX idx_fts_customer_search ON mv_customer_search USING gin (tsv)
如何Column在列甚至存在之前传递可以创建该索引的对象？SqlAlchemy 有一种在建立关系时使用字符串的机制，但这不适用于Index
编辑：尝试使用字符串时添加 Traceback
In [34]: idx = db.Index(
...: "idx_fts_customer_search",
...: "MVCustomerSearch.tsv",
...: postgresql_using="gin",
...: )
In [35]: print(CreateIndex(idx).compile(dialect=postgresql.dialect()))
---------------------------------------------------------------------------
CompileError Traceback (most recent call last)
<ipython-input-35-00025f58e3f9> in <module>
----> 1 print(CreateIndex(idx).compile(dialect=postgresql.dialect()))
<string> in <lambda>(self, bind, dialect, **kw)
~\AppData\Local\pypoetry\Cache\virtualenvs\pce-testsheets-g8TS-oaq-py3.8\Lib\site-packages\sqlalchemy\sql\elements.py in compile(self, default, bind, dialect, **kw)
479 else:
480 dialect = default.StrCompileDialect()
--> 481 return self._compiler(dialect, bind=bind, **kw)
482
483 def _compiler(self, dialect, **kw):
~\AppData\Local\pypoetry\Cache\virtualenvs\pce-testsheets-g8TS-oaq-py3.8\Lib\site-packages\sqlalchemy\sql\ddl.py in _compiler(self, dialect, **kw)
27 Dialect."""
28
---> 29 return dialect.ddl_compiler(dialect, self, **kw)
30
31
~\AppData\Local\pypoetry\Cache\virtualenvs\pce-testsheets-g8TS-oaq-py3.8\Lib\site-packages\sqlalchemy\sql\compiler.py in __init__(self, dialect, statement, bind, schema_translate_map, compile_kwargs)
320 if self.can_execute:
321 self.execution_options = statement._execution_options
--> 322 self.string = self.process(self.statement, **compile_kwargs)
323
324 @util.deprecated(
~\AppData\Local\pypoetry\Cache\virtualenvs\pce-testsheets-g8TS-oaq-py3.8\Lib\site-packages\sqlalchemy\sql\compiler.py in process(self, obj, **kwargs)
350
351 def process(self, obj, **kwargs):
--> 352 return obj._compiler_dispatch(self, **kwargs)
353
354 def __str__(self):
~\AppData\Local\pypoetry\Cache\virtualenvs\pce-testsheets-g8TS-oaq-py3.8\Lib\site-packages\sqlalchemy\sql\visitors.py in _compiler_dispatch(self, visitor, **kw)
94 )
95 else:
---> 96 return meth(self, **kw)
97
98 else:
~\AppData\Local\pypoetry\Cache\virtualenvs\pce-testsheets-g8TS-oaq-py3.8\Lib\site-packages\sqlalchemy\dialects\postgresql\base.py in visit_create_index(self, create)
2079 preparer = self.preparer
2080 index = create.element
-> 2081 self._verify_index_table(index)
2082 text = "CREATE "
2083 if index.unique:
~\AppData\Local\pypoetry\Cache\virtualenvs\pce-testsheets-g8TS-oaq-py3.8\Lib\site-packages\sqlalchemy\sql\compiler.py in _verify_index_table(self, index)
2993 def _verify_index_table(self, index):
2994 if index.table is None:
-> 2995 raise exc.CompileError(
2996 "Index '%s' is not associated " "with any table." % index.name
2997 )
CompileError: Index 'idx_fts_customer_search' is not associated with any table.