我的数据库中有以下模型:
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)
会议结束后我的
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}")