-
Notifications
You must be signed in to change notification settings - Fork 82
Description
Describe the bug
When you type in a numerical value 0 to filters, it is being sent to backend as empty string "". Later on when sqlalchemy tries to compare numerical value in database to the empty string, the exception is raised: sqlalchemy.exc.DataError: (psycopg2.errors.InvalidTextRepresentation) invalid input syntax for type integer: ""

is sent as http://127.0.0.1:8000/admin/api/numbers?skip=0&limit=20&order_by=id desc&where={"and":[{"value":{"gt":""}}]}
To Reproduce
To reproduce you'd need a PostgreSQL database you can connect to.
Here is complete python script that could show this issue:
from fastapi import FastAPI
from sqlalchemy import Column, Integer, create_engine, text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from starlette_admin.contrib.sqla import Admin, ModelView
from starlette.middleware.sessions import SessionMiddleware
import uvicorn
DATABASE_URL = f"postgresql://postgres:postgres@localhost:5432"
DB_NAME = "starlette_poc"
engine = create_engine(f"{DATABASE_URL}/{DB_NAME}")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
class Numbers(Base):
__tablename__ = "numbers"
id = Column(Integer, primary_key=True, index=True)
value = Column(Integer, nullable=False)
def __str__(self):
return f"Number #{self.id}: {self.value}"
def create_database_if_not_exists():
"""Create the database if it doesn't exist"""
admin_url = f"{DATABASE_URL}/postgres"
admin_engine = create_engine(admin_url)
try:
with admin_engine.connect() as conn:
# Check if database exists
result = conn.execute(text(f"SELECT 1 FROM pg_database WHERE datname = '{DB_NAME}'"))
exists = result.fetchone()
if not exists:
conn.execute(text("COMMIT"))
conn.execute(text(f"CREATE DATABASE {DB_NAME}"))
print(f"Database '{DB_NAME}' created successfully")
else:
print(f"Database '{DB_NAME}' already exists")
except Exception as e:
print(f"Error with database setup: {e}")
print("Please ensure PostgreSQL is running and accessible")
raise
finally:
admin_engine.dispose()
# Create database and tables
create_database_if_not_exists()
Base.metadata.create_all(bind=engine)
# FastAPI app
app = FastAPI(
title="FastAPI + SQLAlchemy + Starlette-Admin Demo (PostgreSQL)",
description="Demo application with Numbers model CRUD operations using PostgreSQL"
)
# Add session middleware (required for starlette-admin)
app.add_middleware(SessionMiddleware, secret_key="your-secret-key-here")
# Admin setup
admin = Admin(engine, title="Numbers Admin")
# Define the ModelView for Numbers
class NumbersAdmin(ModelView):
fields = ["id", "value"]
fields_default_sort = [("id", True)]
page_size = 20
page_size_options = [10, 20, 50, 100]
# Enable all CRUD operations
can_create = lambda *_: True
can_edit = lambda *_: True
can_delete = lambda *_: True
can_view_details = lambda *_: True
# Add the model view to admin
admin.add_view(NumbersAdmin(Numbers))
# Mount the admin interface
admin.mount_to(app)
if __name__ == "__main__":
# Add some sample data on startup
db = SessionLocal()
try:
# Check if we already have data
count = db.query(Numbers).count()
if count == 0:
# Add sample numbers
sample_numbers = [
Numbers(value=42),
Numbers(value=100),
Numbers(value=7),
Numbers(value=999),
Numbers(value=13)
]
db.add_all(sample_numbers)
db.commit()
print("Sample data added to database")
finally:
db.close()
print("Starting FastAPI + Starlette-Admin Demo with PostgreSQL")
uvicorn.run(app, host="0.0.0.0", port=8000)
Environment (please complete the following information):
- Starlette-Admin == "0.15.1"
- SQLAlchemy == "2.0.43"
- fastapi == "0.116.1"
- postgresql == 16.2
Additional context
If you swap database for SQLite, it works fine.
I also have a backend workaround for that. I don't really like it (it's ugly), but it fixes the issue by replacing "" with 0 for a specific filter.
You need to add following code to NumbersAdmin (or your own Admin) class and rename the value
to your column name. If you have multiple integer columns, you might need to modify this code accordingly
async def find_all(
self,
request: Request,
skip: int = 0,
limit: int = 100,
where: Union[dict[str, Any], str, None] = None,
order_by: Optional[list[str]] = None,
) -> Sequence[Any]:
return await super().find_all(
request=request,
skip=skip,
limit=limit,
where=self._fix_where(where)
if "value" in str(request.query_params)
else where,
order_by=order_by,
)
async def count(
self,
request: Request,
where: Union[dict[str, Any], str, None] = None,
) -> int:
return await super().count(
request=request,
where=self._fix_where(where)
if "value" in str(request.query_params)
else where,
)
@staticmethod
def _fix_where(where: Union[dict[str, Any], str, None] = None) -> dict:
json_filter = json.dumps(where)
if "value" not in json_filter:
return where
pattern = r"(\"value\": {\"\w+\": )\"\"(\})"
new_filter = re.sub(pattern, r"\1 0\2", json_filter)
return json.loads(new_filter)