Using TSVector in database model raises NotImplementedError

[arawxx]

If using a TSVector type column (for Full-text Search purposes in PostgreSQL) in the database model raises a NotImplementedError. I think one could simply ignore this column in the admin panel?

Edit: I'm using the TSVectorType from the SQLAlchemy-Utils package.
Edit 2: I suggest we add a field such as "ignore_fields" which takes in a set / list of strings and does not add them to the view inside the admin panel.

[igorbenav]

Good one, I didn't reallt think about this. Can you show me a more detailed example to reproduce?

[arawxx]

Sure, @YousefAldabbas seems to have PR-ed a solution for this, but here it is anyways.

A sample model from my application:

from sqlalchemy import (
    Column,
    String,
    DateTime,
    ForeignKey,
    Enum,
    Integer,
    Index,
    CheckConstraint,
)
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship, Mapped, mapped_column
from sqlalchemy_utils.types import TSVectorType

from app.models.base import BaseModel
from app.models.another_model import AnotherModel


class MyModel(BaseModel):
    """A SQLAlchemy model for my application's database."""

    __tablename__ = "my_model"
    __table_args__ = ...

    field_1: Mapped[UUID4] = mapped_column(
        UUID(as_uuid=True), ForeignKey("another_model.id"), nullable=False, index=True
    )
    field_2: Mapped[AwareDatetime] = mapped_column(
        DateTime(timezone=True), nullable=False, index=True
    )
    ...
    ...
    ...
    field_7: Mapped[str] = mapped_column(String(1000), nullable=True)
    search_vector = Column(TSVectorType("notes"))

    # Relationships
    another_model = relationship("AnotherModel", back_populates="my_model")

    def __repr__(self):
        return f"<MyModel {self.field_1} - {self.field_2}>"

Adding it to CRUDAdmin:

...
admin.add_view(
    model=MyModel,
    create_schema=MyModelCreate,
    update_schema=MyModelUpdate,
    allowed_actions={"view", "create", "update", "delete"},
    update_internal_schema=None,
    delete_schema=None,
)
...

Running my FastAPI app raises:

.../.venv/lib/python3.13/site-packages/fastcrud/endpoint/helper.py:117 in _get_python_type

  114 │   │   if _is_uuid_type(column.type):                          ╭──────────────────────────────── locals ────────────────────────────╮
  115 │   │   │   return UUID                                         │ column = Column('search_vector', TSVectorType(), table=<my_model>) │
  116 │   │                                                           ╰────────────────────────────────────────────────────────────────────╯
❱ 117 │   │   direct_type: Optional[type] = column.type.python_type                                                                             
  118 │   │   return direct_type                                                                                                                
  119 │   except NotImplementedError:                                                                                                           
  120 │   │   if hasattr(column.type, "impl") and hasattr(column.type.impl, "python_type"):                                                     
                                                                        
.../.venv/lib/python3.13/site-packages/sqlalchemy/sql/type_api.py:643 in python_type                                
                                                                        
   640 │   │   in practice.                                           ╭─────── locals ────────╮                                                
   641 │   │                                                          │ self = TSVectorType() │                                                
   642 │   │   """                                                    ╰───────────────────────╯                                                
❱  643 │   │   raise NotImplementedError()                                                                                                      
   644 │                                                                                                                                        
   645 │   def with_variant(                                                                                                                    
   646 │   │   self,
────────────────────────────────────────────────────────────────────────────────────────────────────────────────
NotImplementedError

[arab0v]

After digging more into this issue in sqlalchemy-utils repo I find out they have fixed it while updating the package to support latest update for sqlalchemy, or python_type introduced in sqlalchemy 2 (that what I think)
kvesteri/sqlalchemy-utils#733

[arawxx]

I updated the dependencies and the CRUDAdmin all to their latest versions, but the problem still persists.

[arab0v]

This issue is very confusing, I was running wrong tests while debugging in sqlalchemy-utils repo. from what I understood TsVectorType is a wrapper over TSVECTOR which is not a column type and wasn't meant to be used as one, sqlalchemy-utils uses it as column type but not like the other "native" sqlalchemy types it doesn't include some property which python_type is one of them.
issue could be solved by adding python_type to TSVectorType class

So @arawxx to continue your testing you can create new class inherated from TsVectorType and add python_type property

from sqlalchemy_utils.types import TSVectorType as TSVectorTypeBase

class TSVectorType(TSVectorTypeBase):
    @property
    def python_type(self) -> type[str]:
        return str


Base = declarative_base()


class Article(Base):
    __tablename__ = "article"

    id = Column(Integer, primary_key=True)
    name = Column(String(255))
    content = Column(Text)
    search_vector = Column(TSVectorType("name", "content"), nullable=True, )

and you will be able to run the app

Honestly I think this is an issue from slqalchemy-utils or at least that what I understood, I will open a PR there and see whats happen.

@igorbenav let me know what you think.

minimal code to reproduce and see the fix

from contextlib import asynccontextmanager
from typing import AsyncGenerator, Optional

from fastapi import FastAPI, Depends
from sqlalchemy import Column, DateTime, Integer, String, Text
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import Mapped, mapped_column, configure_mappers, declarative_base
from sqlalchemy_searchable import make_searchable

DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/testdb"
engine = create_async_engine(DATABASE_URL)

async_engine = create_async_engine(DATABASE_URL, echo=False, future=True)
local_session = async_sessionmaker(
    bind=async_engine, class_=AsyncSession, expire_on_commit=False
)


from sqlalchemy_utils.types import TSVectorType as TSVectorTypeBase

class TSVectorType(TSVectorTypeBase):
    @property
    def python_type(self) -> type[str]:
        return str


Base = declarative_base()


class Article(Base):
    __tablename__ = "article"

    id = Column(Integer, primary_key=True)
    name = Column(String(255))
    content = Column(Text)
    search_vector = Column(TSVectorType("name", "content"), nullable=True, )


from crudadmin import CRUDAdmin

app = FastAPI()


# Create database session dependency
async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with local_session() as session:
        yield session


# # Create admin interface
admin = CRUDAdmin(
    session=get_session,
    SECRET_KEY="your-secret-key-here",
    initial_admin={"username": "admin", "password": "secure_password123"},
    secure_cookies=False,
)


from pydantic import BaseModel, Field


class ArticleCreate(BaseModel):
    name: str = Field(..., max_length=255)
    content: str


class ArticleUpdate(BaseModel):
    name: Optional[str] = Field(..., max_length=255)
    content: Optional[str]









# Add models to admin
admin.add_view(
    model=Article,
    create_schema=ArticleCreate,
    update_schema=ArticleUpdate,
    allowed_actions={"view", "create", "update"},
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialize database tables
    async with engine.begin() as conn:
        configure_mappers()
        await conn.run_sync(Base.metadata.create_all)

    # Initialize admin interface
    await admin.initialize()
    yield


# Create and mount the app
app = FastAPI(lifespan=lifespan)
app.mount("/admin", admin.app)



@app.post("/articles/", )
async def create_article(article: ArticleCreate, session: AsyncSession = Depends(get_session)):
    new_article = Article(name=article.name, content=article.content)
    session.add(new_article)
    await session.commit()
    
    
    return new_article


@app.get("/articles/{article_id}", )
async def read_article(article_id: int, session: AsyncSession = Depends(get_session)):
    article = await session.get(Article, article_id)
    if article is None:
        raise ValueError("Article not found")
    return article


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

[igorbenav]

@arab0v great assessment! I think we can just add a note about this in the docs until sqlalchemy-utils fix this, then we can close this issue. Do you want to do it?

[arab0v]

@igorbenav Will add it to fastcrud since its being raised from there