SQLAlchemy 2.0 查询获取每个主题的最近(按日期)访问

问题描述 投票:0回答:1

我的数据库中有以下模型:

class Subject(Base):
    __tablename__ = 'subjects'

    id: Mapped[int] = mapped_column(primary_key=True)
    first_name: Mapped[str] = mapped_column(String(60), nullable=False)
    last_name: Mapped[str] = mapped_column(String(60), nullable=False)
    visits: Mapped[list['Visit']] = relationship(cascade='all, delete-orphan',
                                                 back_populates='subject')

class Visit(Base):
    __tablename__ = 'visits'

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    date: Mapped[str] = mapped_column(DateTime, nullable=False)
    amount_spent: Mapped[int] = mapped_column(Integer, nullable=False)
    units: Mapped[str] = mapped_column(String, nullable=False)
    subject_id: Mapped[int] = mapped_column(Integer, ForeignKey('subjects.id'), index=True)
    subject: Mapped['Subject'] = relationship(back_populates='visits')

我现在想编写一个查询,返回数据库中每个主题的最近访问情况。

这是我到目前为止所拥有的:

q = select(Visit) \
            .join(Subject.visits) \
            .order_by(Visit.date.desc()).limit(1)

但是当我尝试打印/访问结果时,运行此查询会导致

DetachedInstanceError
错误消息。

编辑:这是我运行的代码:

q = select(Visit) \
            .join(Subject.visits) \
            .order_by(Visit.date.desc()).limit(1)
r = session.scalars(q).all()
r[0]

这是完整的错误消息:

DetachedInstanceError                     Traceback (most recent call last)
File ~\Anaconda3\envs\cam\Lib\site-packages\IPython\core\formatters.py:708, in PlainTextFormatter.__call__(self, obj)
    701 stream = StringIO()
    702 printer = pretty.RepresentationPrinter(stream, self.verbose,
    703     self.max_width, self.newline,
    704     max_seq_length=self.max_seq_length,
    705     singleton_pprinters=self.singleton_printers,
    706     type_pprinters=self.type_printers,
    707     deferred_pprinters=self.deferred_printers)
--> 708 printer.pretty(obj)
    709 printer.flush()
    710 return stream.getvalue()

File ~\Anaconda3\envs\cam\Lib\site-packages\IPython\lib\pretty.py:410, in RepresentationPrinter.pretty(self, obj)
    407                         return meth(obj, self, cycle)
    408                 if cls is not object \
    409                         and callable(cls.__dict__.get('__repr__')):
--> 410                     return _repr_pprint(obj, self, cycle)
    412     return _default_pprint(obj, self, cycle)
    413 finally:

File ~\Anaconda3\envs\cam\Lib\site-packages\IPython\lib\pretty.py:778, in _repr_pprint(obj, p, cycle)
    776 """A pprint that just redirects to the normal repr function."""
    777 # Find newlines and replace them with p.break_()
--> 778 output = repr(obj)
    779 lines = output.splitlines()
    780 with p.group():

File ~\Anaconda3\envs\cam\Lib\site-packages\sqlalchemy\engine\row.py:245, in Row.__repr__(self)
    244 def __repr__(self) -> str:
--> 245     return repr(sql_util._repr_row(self))

File ~\Anaconda3\envs\cam\Lib\site-packages\sqlalchemy\sql\util.py:598, in _repr_row.__repr__(self)
    595 def __repr__(self) -> str:
    596     trunc = self.trunc
    597     return "(%s%s)" % (
--> 598         ", ".join(trunc(value) for value in self.row),
    599         "," if len(self.row) == 1 else "",
    600     )

File ~\Anaconda3\envs\cam\Lib\site-packages\sqlalchemy\sql\util.py:598, in <genexpr>(.0)
    595 def __repr__(self) -> str:
    596     trunc = self.trunc
    597     return "(%s%s)" % (
--> 598         ", ".join(trunc(value) for value in self.row),
    599         "," if len(self.row) == 1 else "",
    600     )

File ~\Anaconda3\envs\cam\Lib\site-packages\sqlalchemy\sql\util.py:565, in _repr_base.trunc(self, value)
    564 def trunc(self, value: Any) -> str:
--> 565     rep = repr(value)
    566     lenrep = len(rep)
    567     if lenrep > self.max_chars:

File models.py:93, in Visit.__repr__(self)
      92 def __repr__(self):
 ---> 93     return f"{self.date.strftime('%Y-%m-%d')}"

File ~\Anaconda3\envs\cam\Lib\site-packages\sqlalchemy\orm\attributes.py:566, in InstrumentedAttribute.__get__(self, instance, owner)
    564 except AttributeError as err:
    565     raise orm_exc.UnmappedInstanceError(instance) from err
--> 566 return self.impl.get(state, dict_)

File ~\Anaconda3\envs\cam\Lib\site-packages\sqlalchemy\orm\attributes.py:1086, in AttributeImpl.get(self, state, dict_, passive)
   1083 if not passive & CALLABLES_OK:
   1084     return PASSIVE_NO_RESULT
-> 1086 value = self._fire_loader_callables(state, key, passive)
   1088 if value is PASSIVE_NO_RESULT or value is NO_VALUE:
   1089     return value

File ~\Anaconda3\envs\cam\Lib\site-packages\sqlalchemy \orm\attributes.py:1116, in AttributeImpl._fire_loader_callables(self, state, key, passive)
   1108 def _fire_loader_callables(
   1109     self, state: InstanceState[Any], key: str, passive:  PassiveFlag
   1110 ) -> Any:
   1111     if (
   1112         self.accepts_scalar_loader
   1113         and self.load_on_unexpire
   1114         and key in state.expired_attributes
   1115     ):
-> 1116         return state._load_expired(state, passive)
   1117     elif key in state.callables:
   1118         callable_ = state.callables[key]

File ~\Anaconda3\envs\cam\Lib\site-packages\sqlalchemy\orm\state.py:798, in InstanceState._load_expired(self, state, passive)
    791 toload = self.expired_attributes.intersection(self.unmodified)
    792 toload = toload.difference(
    793     attr
    794     for attr in toload
    795     if not self.manager[attr].impl.load_on_unexpire
    796 )
--> 798 self.manager.expired_attribute_loader(self, toload, passive)
    800 # if the loader failed, or this
    801 # instance state didn't have an identity,
    802 # the attributes still might be in the callables
    803 # dict.  ensure they are removed.
    804 self.expired_attributes.clear()

File ~\Anaconda3\envs\cam\Lib\site-packages\sqlalchemy\orm\loading.py:1582, in load_scalar_attributes(mapper, state, attribute_names, passive)
   1580 session = state.session
   1581 if not session:
-> 1582     raise orm_exc.DetachedInstanceError(
   1583         "Instance %s is not bound to a Session; "
   1584         "attribute refresh operation cannot proceed" % (state_str(state))
   1585     )
   1587 no_autoflush = bool(passive & attributes.NO_AUTOFLUSH)
   1589 # in the case of inheritance, particularly concrete and abstract
   1590 # concrete inheritance, the class manager might have some keys
   1591 # of attributes on the superclass that we didn't actually map.
   1592 # These could be mapped as "concrete, don't load" or could be completely
   1593 # excluded from the mapping and we know nothing about them.  Filter them
   1594 # here to prevent them from coming through.

DetachedInstanceError: Instance <Visit at 0x166c3b74250> is not bound to a Session; attribute refresh operation cannot proceed (Background on this error at: https://sqlalche.me/e/20/bhk3)
python python-3.x sqlite sqlalchemy
1个回答
0
投票

会议结束后我的

print
加注
sqlalchemy.orm.exc.DetachedInstanceError

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


Base = declarative_base()


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


uri = f"postgresql+psycopg2://{username}:{password}@/{db}"


engine = create_engine(uri, echo=False)


class Post(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True, nullable=False)
    title = Column(String, nullable=False)


Base.metadata.create_all(engine)

p1 = Post(title="The Next Big Thing")

with Session(engine) as session, session.begin():
    session.add(p1)
    session.commit()

print(f"{p1.title}")
© www.soinside.com 2019 - 2024. All rights reserved.