postgresql AttributeError(“与Symbols.symbol关联的“InstrumentedAttribute”对象和“Comparator”对象都没有属性“split”“)

lp0sw83n  于 2023-06-29  发布在  PostgreSQL
关注(0)|答案(1)|浏览(115)

我试图从一个包含 * 股票代码 * 沿着相关数据的sql表中选择行。
然而,表中的一些行具有以下格式的符号{symbol}.{exchange} for example,AAPL.NQ。 我试图查找所有行,将匹配一个特定的股票代码'ZYX'不管尾随交易所。所以对于属于的例子,symbol_ = 'ZYX'包含符号ZYXZYX.AB',` ZYX.DB'的所有行都将被提取。

symbol_ = 'ZYX'
currency_ = 'USD'
security_type_ = 'etf'

matching_pi = session.execute(select(Symbols).where(Symbols.symbol.split('.')[0] == symbol_)).scalars().all()

添加.split('.')[0] is resulting in the following error which is expected. AttributeError(“与Symbols.symbol关联的'InstrumentedAttribute'对象和'Comparator'对象都没有属性'split'")我想忽略符号名称中.之后的符号字符串部分。ABC.XZ应解释为ABC`

5cg8jx4n

5cg8jx4n1#

你可以使用startswith。

from sqlalchemy.sql import or_

symbol_ = 'ZYX'
currency_ = 'USD'
security_type_ = 'etf'

matching_pi = session.execute(select(Symbols).where(or_(Symbols.symbol == symbol_, Symbols.symbol.startswith(symbol_ + ".")))

完整示例

import sys
from sqlalchemy import (
    create_engine,
    Integer,
    String,
)
from sqlalchemy.schema import (
    Column,
)
from sqlalchemy.orm import Session, declarative_base
from sqlalchemy.sql import select, or_, func

username, password, db = sys.argv[1:4]

engine = create_engine(f"postgresql+psycopg2://{username}:{password}@/{db}", echo=True)

Base = declarative_base()

class Symbol(Base):
    __tablename__ = 'symbols'
    id = Column(Integer, primary_key=True)
    symbol = Column(String)

Base.metadata.create_all(engine)


with Session(engine) as session:
    # Insert valid and invalid symbols.
    valid_labels = ["abc", "abc.def"]
    invalid_labels = ["abcx", "abcx.def", "ab.c"]
    session.add_all([Symbol(symbol=s) for s in valid_labels] + [Symbol(symbol=s) for s in invalid_labels])
    session.commit()

with Session(engine) as session:
    # Make sure all symbols were inserted.
    stmt = select(Symbol)
    symbols = session.execute(stmt).scalars().all()
    assert len(symbols) == len(valid_labels + invalid_labels)

def test_labels(found_labels):
    # Make sure the found labels are the valid ones.
    assert len(found_labels) == len(valid_labels)
    for label in valid_labels:
        assert label in found_labels
    for label in invalid_labels:
        assert label not in found_labels

with Session(engine) as session:
    # Lookup symbols matching our test symbol with split_part
    symbol = 'abc'
    stmt = select(Symbol).where(func.split_part(Symbol.symbol, ".", 1) == symbol)
    symbols = session.execute(stmt).scalars().all()
    found_labels = [s.symbol for s in symbols]

    test_labels(found_labels)

with Session(engine) as session:
    # Lookup symbols matching our test symbol with or(equals,startswith)
    symbol = 'abc'
    stmt = select(Symbol).where(or_(Symbol.symbol == symbol, Symbol.symbol.startswith(symbol + ".")))
    symbols = session.execute(stmt).scalars().all()
    found_labels = [s.symbol for s in symbols]

    test_labels(found_labels)

相关问题