Skip to content
Advertisement

Writing a hybrid method using pathlib (accessing properties)

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.

Advertisement

Answer

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(?, '/') || '/' || '%'
)
Advertisement