to the value ``"raise_on_sql"``, so that for a particular mapping, a certain
relationship will never try to emit SQL:
-.. sourcecode:: python
-
- from sqlalchemy.orm import Mapped
- from sqlalchemy.orm import relationship
-
+.. setup code
- class User(Base):
- __tablename__ = "user_account"
+ >>> class Base(DeclarativeBase):
+ ... pass
- # ... mapped_column() mappings
+::
- addresses: Mapped[list["Address"]] = relationship(
- back_populates="user", lazy="raise_on_sql"
- )
+ >>> from sqlalchemy.orm import Mapped
+ >>> from sqlalchemy.orm import relationship
- class Address(Base):
- __tablename__ = "address"
+ >>> class User(Base):
+ ... __tablename__ = "user_account"
+ ... id: Mapped[int] = mapped_column(primary_key=True)
+ ... addresses: Mapped[list["Address"]] = relationship(
+ ... back_populates="user", lazy="raise_on_sql"
+ ... )
- # ... mapped_column() mappings
- user: Mapped["User"] = relationship(back_populates="addresses", lazy="raise_on_sql")
+ >>> class Address(Base):
+ ... __tablename__ = "address"
+ ... id: Mapped[int] = mapped_column(primary_key=True)
+ ... user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
+ ... user: Mapped["User"] = relationship(back_populates="addresses", lazy="raise_on_sql")
Using such a mapping, the application is blocked from lazy loading,
-indicating that a particular query would need to specify a loader strategy:
-
-.. sourcecode:: python
+indicating that a particular query would need to specify a loader strategy::
- u1 = s.execute(select(User)).scalars().first()
- u1.addresses
+ >>> u1 = session.execute(select(User)).scalars().first()
+ {opensql}SELECT user_account.id FROM user_account
+ [...] ()
+ {stop}>>> u1.addresses
+ Traceback (most recent call last):
+ ...
sqlalchemy.exc.InvalidRequestError: 'User.addresses' is not available due to lazy='raise_on_sql'
The exception would indicate that this collection should be loaded up front
-instead:
+instead::
-.. sourcecode:: python
-
- u1 = s.execute(select(User).options(selectinload(User.addresses))).scalars().first()
+ >>> u1 = (
+ ... session.execute(select(User).options(selectinload(User.addresses)))
+ ... .scalars()
+ ... .first()
+ ... )
+ {opensql}SELECT user_account.id
+ FROM user_account
+ [...] ()
+ SELECT address.user_id AS address_user_id, address.id AS address_id
+ FROM address
+ WHERE address.user_id IN (?, ?, ?, ?, ?, ?)
+ [...] (1, 2, 3, 4, 5, 6)
The ``lazy="raise_on_sql"`` option tries to be smart about many-to-one
relationships as well; above, if the ``Address.user`` attribute of an
config.skip_test("Can't find documentation file %r" % path)
buf = []
- line_counter = 0
- last_line_counter = 0
+
with open(path, encoding="utf-8") as file_:
def load_include(m):
fname = m.group(1)
sub_path = os.path.join(os.path.dirname(path), fname)
with open(sub_path, encoding="utf-8") as file_:
- for line in file_:
- buf.append(line)
+ for i, line in enumerate(file_, 1):
+ buf.append((i, line))
return fname
def run_buf(fname, is_include):
if not buf:
return
- nonlocal last_line_counter
+
test = parser.get_doctest(
- "".join(buf),
+ "".join(line for _, line in buf),
globs,
fname,
fname,
- last_line_counter if not is_include else 0,
+ buf[0][0],
)
buf[:] = []
runner.run(test, clear_globs=False)
globs.update(test.globs)
- if not is_include:
- last_line_counter = line_counter
+ doctest_enabled = True
- for line in file_:
+ for line_counter, line in enumerate(file_, 1):
line = re.sub(r"{(?:stop|sql|opensql)}", "", line)
include = re.match(r"\.\. doctest-include (.+\.rst)", line)
run_buf(fname, False)
include_fname = load_include(include)
run_buf(include_fname, True)
+
+ doctest_disable = re.match(
+ r"\.\. doctest-(enable|disable)", line
+ )
+ if doctest_disable:
+ doctest_enabled = doctest_disable.group(1) == "enable"
+
+ if doctest_enabled:
+ buf.append((line_counter, line))
else:
- buf.append(line)
- line_counter += 1
+ buf.append((line_counter, "\n"))
run_buf(fname, False)