使用pathlib实现SQLAlchemy混合方法的路径检查逻辑问题

需要对数据库中的路径值执行路径检查逻辑，判断给定路径是否是记录中路径的自身或父路径，但在实现SQLAlchemy的hybrid_method类级表达式时遇到困难，运行时报错：Neither 'Function' object nor 'Comparator' object has an attribute 'parents'。

精简后的代码如下：

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()

核心问题是不能在SQL表达式中直接使用Python的Path对象及其parents属性——SQLite无法识别Python对象的属性操作，必须把路径检查逻辑转换成SQL原生的字符串操作：

判断给定路径a是否是记录路径b的自身或父路径，等价于：

1. b以a开头；
2. 要么b完全等于a，要么b在a之后紧跟着路径分隔符/（避免误匹配类似/folder1和/folder12的情况）。

修改后的代码如下：

from sqlalchemy import Column, Integer, String, create_engine, func, select
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.ext.hybrid import 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):
        # 构造SQL层面的路径匹配条件
        # 1. 记录路径等于给定路径
        # 2. 记录路径以给定路径 + '/' 开头
        return sa.or_(
            cls.fpath == fpath,
            cls.fpath.startswith(f"{fpath}/")
        )


def db_create_row(db: Session, value: int, fpath: str):
    folder = Folder(value=value, fpath=fpath)
    db.add(folder)
    db.commit()
    db.refresh(folder)


# 判断给定路径是否是表中任意记录的父路径（或自身）
def db_query_row(db: Session, fpath: str):
    records = db.query(Folder).filter(Folder.test_path(fpath))
    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())  # 输出1，符合预期
    db.close()

• 移除了无用的Path函数注册：SQLite无法处理Python的Path对象，注册这个函数没有意义；
• 重写test_path.expression：用SQL原生的==和startswith实现路径匹配逻辑，完全适配SQL层面的查询；
• 简化查询条件：filter(Folder.test_path(fpath))即可，无需额外判断== True。

内容的提问来源于stack exchange，提问作者Mr. Anderson