I'm trying to do some path-checking logic on some values in a database. I'm having trouble implementing the class level expression for the hybrid method.
Here is some stripped down code:
from sqlalchemy import Column, Integer, String, create_engine, func, select
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session, aliased, sessionmaker
from sqlalchemy.ext.hybrid import hybrid_property, hybrid_method
from pathlib import Path
import sqlalchemy as sa
dbEngine = create_engine(
"sqlite:///.sql_app.db", connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=dbEngine)
Base = declarative_base()
class Folder(Base):
__tablename__ = "folder"
id = Column(Integer, primary_key=True)
value = Column(Integer, nullable=False)
fpath = Column(String, index=True)
@hybrid_method
def test_path(self, fpath):
a = Path(fpath)
b = Path(self.fpath)
return a in [b] [p for p in b.parents]
@test_path.expression
def test_path(cls, fpath):
a = func.Path(fpath)
b = func.Path(cls.fpath)
# return a in [b] [p for p in b.parents]
# What to do here
return (
select([func.Path(Folder.fpath)]).
label("fpath_in_folder")
)
@sa.event.listens_for(sa.engine.Engine, "connect")
def sqlite_engine_connect(dbapi_conn, connection_record) -> None:
dbapi_conn.create_function("Path", 1, Path)
def db_create_row(db: Session, value: int, fpath: str):
folder = Folder(value=value, fpath=fpath)
db.add(folder)
db.commit()
db.refresh(folder)
# Determine if the supplied path is a sub path of any records in the table
def db_query_row(db: Session, fpath: str):
records = db.query(Folder).filter(Folder.test_path(fpath) == True)
return records
if __name__ == "__main__":
Base.metadata.create_all(bind=dbEngine)
db = SessionLocal()
db_create_row(db, 5, "/folder 1/folder 2/file.ext")
records = db_query_row(db, "/folder 1")
print(records.count())
db.close()
I'm getting the error: Neither 'Function' object nor 'Comparator' object has an attribute 'parents'
So I have to create a SQL expression for this function, but I'm not sure how, or if it's even possible with accessing the parents
property on the Path object.
CodePudding user response:
SQLite can't handle Python instances, and SQLAlchemy Function
support doesn't cover instance methods either.
First of all, you can't use Path()
objects as a SQLite function, as explained in the sqlite3.Connection.create_function
documentation:
The callable must return a type natively supported by SQLite.
The natively supported types are None
, float
, int
, str
or bytes
values.
The error you see comes from your attempt to use b.parents
in your expression; b
is the Function
object, the func.Path(...)
call, and SQLAlchemy expects a function to return a SQL type, not a Path()
object.
Instead of trying to shoehorn Path()
objects into SQLite, you'll need to find another way to test if a path is a parent folder. You could use Column.startswith()
here, provided you first ensure the paths don't end with /
, by using the standard RTRIM()
function:
@test_path.expression
def test_path(cls, fpath):
a = func.rtrim(fpath, "/")
b = func.rtrim(cls.fpath, "/")
return a == b or b.startswith(a "/")
This will produce a SQL expression like this:
(
rtrim(?, '/') == rtrim(folder.fpath, '/')
OR rtrim(folder.fpath, '/') LIKE rtrim(?, '/') || '/' || '%'
)