The easiest way to add OData filtering to a SQLAlchemy query is with the shorthand:
from odata_query.sqlalchemy import apply_odata_query
orm_query = select(MyModel) # This is any form of Query or Selectable.
odata_query = "name eq 'test'" # This will usually come from a query string parameter.
query = apply_odata_query(orm_query, odata_query)
results = session.execute(query).scalars().all()
Attention
Basic support for SQLAlchemy Core is new since version 0.7.0. It currently does not support relationship traversal or any
/all
yet. Those operations will raise a NotImplementedException
.
from odata_query.sqlalchemy import apply_odata_core
core_query = select(MyTable) # This is any form of Query or Selectable.
odata_query = "name eq 'test'" # This will usually come from a query string parameter.
query = apply_odata_query(core_query, odata_query)
results = session.execute(query).scalars().all()
If you need some more flexibility or advanced features, the implementation of the shorthand is a nice starting point: :pyodata_query.sqlalchemy.shorthand
Let's break it down real quick:
To get from an AST
to something SQLAlchemy can use, you'll need to use the :pyodata_query.sqlalchemy.orm.AstToSqlAlchemyOrmVisitor
(ORM mode) or the :pyodata_query.sqlalchemy.core.AstToSqlAlchemyCoreVisitor
(Core mode). It needs to know about the 'root model' or table of your query in order to see which fields exists and how objects are related.
from odata_query.sqlalchemy.orm import AstToSqlAlchemyOrmVisitor
visitor = AstToSqlAlchemyOrmVisitor(MyModel)
query_filter = visitor.visit(ast)
from odata_query.sqlalchemy.core import AstToSqlAlchemyCoreVisitor
visitor = AstToSqlAlchemyCoreVisitor(MyTable)
query_filter = visitor.visit(ast)
Attention
Relationship traversal and automatic joins are not yet supported for SQLAlchemy Core mode.
If your query spans relationships, the AstToSqlAlchemyClauseVisitor
will generate join statements. For the query to work, these will need to be applied explicitly:
for j in visitor.join_relationships:
query = query.join(j)
Finally, we're ready to run the query:
query = query.where(query_filter)
results = s.execute(query).scalars().all()