mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 19:01:39 -04:00
db: serializers read only loaded data (#777)
* db: serializers read only loaded data Serializing a model must not query. A lazy load there is a hidden query per row (search serialized 24 rows with 55 statements), and would be a `MissingGreenlet` error under an async Session. - `wrolpi.db.no_db_access(reason)` and `@serializer` (on every model `__json__`, and around `json_response`): an engine `before_cursor_execute` listener raises `UnexpectedQuery` during tests, and in production logs once per serializer and lets the query run. Tests record each trip as well, and the autouse `no_unexpected_queries` fixture fails a test even when the code caught the error (Video's serializer was swallowing its per-row loads of the deferred `ffprobe_json`). - Relationships serializers read load with their parent: `selectin` for FileGroup.tag_files and Collection.downloads; `joined` for TagFile.tag, Collection.tag, Channel.collection, Video.channel, Archive.collection, ZimSubscription.download, CollectionItem.zim, and the file_group of Video, Archive, Doc, Zim and CollectionItem. - Search and the Video lookups the video page uses load the deferred `ffprobe_json` (`undefer`). - `wrolpi/test/test_serializers.py`: every serialized model, freshly loaded, serializes without SQL. Search: 55 -> 5 statements for 24 videos (31 -> 5 for archives), and a few ms faster locally. Not in this change: guarding the `dict()` serializers and `expire_on_commit=False`, which go together (about 50 serialize-after-commit sites, and 40 tests that read values changed elsewhere after a commit). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * db: serializer loads per query; the guard ignores transaction control The query guard reported the engine's own BEGIN (emitted through the same event as the first statement on a Session with no open transaction) instead of the lazy load after it, and in production that BEGIN used up the one warning per serializer. Transaction-control statements are now ignored. Mapper-level eager loads reached background code: every Video/Archive load in the modelers (one per file, inside the write transaction) joined file_group/channel/collection/tag and selectin-loaded the Collection's Download rows with their yt-dlp info_json. A second refresh of 100 videos took 0.40 s instead of 0.16 s. Serializer needs are now declared per model (`json_options()`) and added only by the queries whose results are serialized; Download.info_json is deferred there (no serializer reads it). Mapper defaults keep only narrow joins nearly every use needs: TagFile.tag, Collection.tag and Channel.collection. A second refresh is back to master's statement count and time; search still drops from 55 to 3 statements. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
lrnselfreliance
GitHub
parent
1f687fa825
commit
cfbed9e4d7
24 changed files with 371 additions and 72 deletions
|
|
@ -5,13 +5,14 @@ from typing import Iterable, List, Optional
|
|||
|
||||
import pytz
|
||||
from sqlalchemy import Column, Integer, ForeignKey, BigInteger, Index
|
||||
from sqlalchemy.orm import relationship, Session
|
||||
from sqlalchemy.orm import relationship, Session, joinedload
|
||||
|
||||
from wrolpi import dates
|
||||
from wrolpi.collections import Collection
|
||||
from wrolpi.common import ModelHelper, Base, logger, get_title_from_html, get_wrolpi_config, get_media_directory, \
|
||||
strip_surrogates
|
||||
from wrolpi.dates import now
|
||||
from wrolpi.db import serializer
|
||||
from wrolpi.errors import UnknownArchive
|
||||
from wrolpi.files.models import FileGroup
|
||||
from wrolpi.tags import TagFile
|
||||
|
|
@ -39,6 +40,13 @@ class Archive(Base, ModelHelper):
|
|||
file_group_id = Column(BigInteger, ForeignKey('file_group.id', ondelete='CASCADE'), unique=True, nullable=False)
|
||||
file_group: FileGroup = relationship('FileGroup')
|
||||
|
||||
@staticmethod
|
||||
def json_options() -> tuple:
|
||||
"""Loader options for a query whose Archives will be serialized: `__json__` reads the FileGroup and its Tags,
|
||||
and the domain (the Collection's name)."""
|
||||
return joinedload(Archive.file_group).selectinload(FileGroup.tag_files), joinedload(Archive.collection)
|
||||
|
||||
@serializer
|
||||
def __json__(self) -> dict:
|
||||
d = self.file_group.__json__()
|
||||
d['archive'] = dict(
|
||||
|
|
|
|||
|
|
@ -340,7 +340,7 @@ def _doc_response(doc) -> dict:
|
|||
|
||||
def _get_doc(session, file_group_id: int):
|
||||
from .models import Doc
|
||||
doc = session.query(Doc).filter_by(file_group_id=file_group_id).one_or_none()
|
||||
doc = session.query(Doc).options(*Doc.json_options()).filter_by(file_group_id=file_group_id).one_or_none()
|
||||
if not doc:
|
||||
raise ValidationError(f'Doc with file_group_id {file_group_id} not found')
|
||||
return doc
|
||||
|
|
@ -375,7 +375,7 @@ def _search_docs(session: Session, search_str=None, author=None, subject=None, l
|
|||
# the much smaller a/b/c columns.
|
||||
match = fts.translate_websearch(search_str, None if deep else fts.ABC_COLUMNS) if search_str else None
|
||||
|
||||
query = session.query(FileGroup).join(Doc, Doc.file_group_id == FileGroup.id)
|
||||
query = session.query(FileGroup).options(*FileGroup.json_options()).join(Doc, Doc.file_group_id == FileGroup.id)
|
||||
|
||||
fts_sq = None
|
||||
if match is not None:
|
||||
|
|
@ -441,7 +441,8 @@ def _search_docs(session: Session, search_str=None, author=None, subject=None, l
|
|||
browse_sql = _doc_browse_sql(order_by if order_by in DOC_BROWSE_ORDERS else 'id')
|
||||
if browse_sql:
|
||||
fg_ids = [i for i, in session.execute(text(browse_sql), dict(limit=limit, offset=offset)).fetchall()]
|
||||
by_id = {fg.id: fg for fg in session.query(FileGroup).filter(FileGroup.id.in_(fg_ids))} if fg_ids else {}
|
||||
by_id = {fg.id: fg for fg in session.query(FileGroup).options(*FileGroup.json_options())
|
||||
.filter(FileGroup.id.in_(fg_ids))} if fg_ids else {}
|
||||
file_groups = [by_id[i] for i in fg_ids]
|
||||
else:
|
||||
file_groups = query.offset(offset).limit(limit).all()
|
||||
|
|
|
|||
|
|
@ -2,10 +2,10 @@ import pathlib
|
|||
from typing import Optional
|
||||
|
||||
from sqlalchemy import Column, Integer, BigInteger, ForeignKey, String, Text, Index, or_
|
||||
from sqlalchemy.orm import relationship, Session
|
||||
from sqlalchemy.orm import relationship, Session, joinedload
|
||||
|
||||
from wrolpi.common import ModelHelper, Base
|
||||
from wrolpi.db import get_db_session
|
||||
from wrolpi.db import get_db_session, serializer
|
||||
from wrolpi.files.models import FileGroup
|
||||
|
||||
EPUB_MIMETYPE = 'application/epub'
|
||||
|
|
@ -65,6 +65,12 @@ class Doc(ModelHelper, Base):
|
|||
path = str(self.file_group.primary_path) if self.file_group else 'None'
|
||||
return f'<Doc id={self.id} path={repr(path)} file_group_id={self.file_group_id}>'
|
||||
|
||||
@staticmethod
|
||||
def json_options() -> tuple:
|
||||
"""Loader options for a query whose Docs will be serialized: `__json__` reads the FileGroup and its Tags."""
|
||||
return joinedload(Doc.file_group).selectinload(FileGroup.tag_files),
|
||||
|
||||
@serializer
|
||||
def __json__(self) -> dict:
|
||||
d = self.file_group.__json__()
|
||||
if d.get('data') and d['data'].get('doc_path'):
|
||||
|
|
|
|||
|
|
@ -208,6 +208,7 @@ async def search_channels_by_name(session: Session, name: str, limit: int = 5,
|
|||
name_order = (func.lower(func.replace(Collection.name, ' ', '')).asc(), Collection.name.desc())
|
||||
if order_by_video_count:
|
||||
stmt = session.query(Channel, func.count(Video.id).label('video_count')) \
|
||||
.options(*Channel.json_options()) \
|
||||
.join(Collection) \
|
||||
.filter(or_(
|
||||
Collection.name.ilike(f'%{name}%'),
|
||||
|
|
@ -220,6 +221,7 @@ async def search_channels_by_name(session: Session, name: str, limit: int = 5,
|
|||
channels = [i[0] for i in stmt]
|
||||
else:
|
||||
stmt = session.query(Channel) \
|
||||
.options(*Channel.json_options()) \
|
||||
.join(Collection) \
|
||||
.filter(or_(
|
||||
Collection.name.ilike(f'%{name}%'),
|
||||
|
|
@ -319,5 +321,6 @@ async def tag_channel(session: Session, tag_name: str | None, directory: pathlib
|
|||
|
||||
async def search_channels(session: Session, tag_names: List[str]) -> List[Channel]:
|
||||
"""Search Tagged Channels."""
|
||||
channels = session.query(Channel).join(Collection).join(Tag).filter(Tag.name.in_(tag_names)).all()
|
||||
channels = session.query(Channel).options(*Channel.json_options()) \
|
||||
.join(Collection).join(Tag).filter(Tag.name.in_(tag_names)).all()
|
||||
return channels
|
||||
|
|
|
|||
|
|
@ -3,14 +3,15 @@ import pathlib
|
|||
from typing import Optional, Dict, List
|
||||
|
||||
from sqlalchemy import Column, Integer, String, Boolean, Date, ForeignKey, BigInteger, Index, text, JSON
|
||||
from sqlalchemy.orm import relationship, Session, deferred
|
||||
from sqlalchemy.orm import relationship, Session, deferred, undefer, joinedload, selectinload
|
||||
from sqlalchemy.orm.collections import InstrumentedList
|
||||
|
||||
from modules.videos.errors import UnknownVideo, UnknownChannel
|
||||
from wrolpi.captions import read_captions, read_captions_with_timestamps
|
||||
from wrolpi.common import Base, ModelHelper, logger, get_media_directory, get_relative_to_media_directory, \
|
||||
background_task
|
||||
from wrolpi.db import get_db_curs, get_db_session, session_curs
|
||||
from wrolpi.db import get_db_curs, get_db_session, session_curs, serializer
|
||||
from wrolpi.collections.models import Collection
|
||||
from wrolpi.downloader import Download
|
||||
from wrolpi.files.lib import split_path_stem_and_suffix
|
||||
from wrolpi.files.worker import file_worker
|
||||
|
|
@ -57,6 +58,14 @@ class Video(ModelHelper, Base):
|
|||
return f'<Video id={self.id} title={repr(self.file_group.title)} path={v} channel={self.channel_id} ' \
|
||||
f'source_id={repr(self.source_id)}>'
|
||||
|
||||
@staticmethod
|
||||
def json_options() -> tuple:
|
||||
"""Loader options for a query whose Videos will be serialized: `__json__` reads the codecs in the deferred
|
||||
ffprobe_json, the FileGroup and its Tags, and the Channel (its Collection loads with it)."""
|
||||
return undefer(Video.ffprobe_json), joinedload(Video.file_group).selectinload(FileGroup.tag_files), \
|
||||
joinedload(Video.channel)
|
||||
|
||||
@serializer
|
||||
def __json__(self) -> dict:
|
||||
d = self.file_group.__json__()
|
||||
|
||||
|
|
@ -233,8 +242,9 @@ class Video(ModelHelper, Base):
|
|||
break
|
||||
|
||||
# Fetch the videos by id, if they exist.
|
||||
previous_video = Video.find_by_id(session, previous_id) if previous_id else None
|
||||
next_video = Video.find_by_id(session, next_id) if next_id else None
|
||||
# The neighbors are serialized next to this Video.
|
||||
previous_video = Video.find_by_id(session, previous_id, Video.json_options()) if previous_id else None
|
||||
next_video = Video.find_by_id(session, next_id, Video.json_options()) if next_id else None
|
||||
|
||||
return previous_video, next_video
|
||||
|
||||
|
|
@ -391,27 +401,27 @@ class Video(ModelHelper, Base):
|
|||
return video
|
||||
|
||||
@staticmethod
|
||||
def get_by_id(session: Session, id_: int) -> Optional['Video']:
|
||||
def get_by_id(session: Session, id_: int, options: tuple = ()) -> Optional['Video']:
|
||||
"""Attempt to find a Video with the provided id. Returns None if it cannot be found."""
|
||||
video = session.query(Video).filter(Video.id == id_).one_or_none()
|
||||
video = session.query(Video).options(*options).filter(Video.id == id_).one_or_none()
|
||||
return video
|
||||
|
||||
@staticmethod
|
||||
def find_by_id(session: Session, id_: int) -> 'Video':
|
||||
def find_by_id(session: Session, id_: int, options: tuple = ()) -> 'Video':
|
||||
"""Find a Video with the provided id, raises an exception if it cannot be found.
|
||||
|
||||
@raise UnknownVideo: if the Video can not be found"""
|
||||
video = Video.get_by_id(session, id_)
|
||||
video = Video.get_by_id(session, id_, options)
|
||||
if not video:
|
||||
raise UnknownVideo(f'Cannot find Video with id {id_}')
|
||||
return video
|
||||
|
||||
@staticmethod
|
||||
def find_by_file_group_id(session: Session, file_group_id: int) -> 'Video':
|
||||
def find_by_file_group_id(session: Session, file_group_id: int, options: tuple = ()) -> 'Video':
|
||||
"""Find a Video by its FileGroup ID, raises an exception if it cannot be found.
|
||||
|
||||
@raise UnknownVideo: if the Video can not be found"""
|
||||
video = session.query(Video).filter(Video.file_group_id == file_group_id).one_or_none()
|
||||
video = session.query(Video).options(*options).filter(Video.file_group_id == file_group_id).one_or_none()
|
||||
if not video:
|
||||
raise UnknownVideo(f'Cannot find Video with file_group_id {file_group_id}')
|
||||
return video
|
||||
|
|
@ -620,7 +630,8 @@ class Channel(ModelHelper, Base):
|
|||
|
||||
videos: InstrumentedList = relationship('Video', primaryjoin='Channel.id==Video.channel_id')
|
||||
collection_id = Column(Integer, ForeignKey('collection.id', ondelete='CASCADE'))
|
||||
collection = relationship('Collection', foreign_keys=[collection_id])
|
||||
# Joined: a Channel's name, directory and Tag live on its Collection.
|
||||
collection = relationship('Collection', foreign_keys=[collection_id], lazy='joined')
|
||||
|
||||
@property
|
||||
def downloads(self) -> list:
|
||||
|
|
@ -938,6 +949,13 @@ class Channel(ModelHelper, Base):
|
|||
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def json_options() -> tuple:
|
||||
"""Loader options for a query whose Channels will be serialized: `__json__` lists the Collection's
|
||||
Downloads (without their yt-dlp info_json, which is not serialized)."""
|
||||
return joinedload(Channel.collection).selectinload(Collection.downloads).defer(Download.info_json),
|
||||
|
||||
@serializer
|
||||
def __json__(self) -> dict:
|
||||
d = dict(
|
||||
directory=self.directory,
|
||||
|
|
|
|||
|
|
@ -127,6 +127,9 @@ async def test_video_channel_refresh(async_client, test_session, test_directory,
|
|||
video2: Video = Video.get_by_path(test_session, video2_path)
|
||||
assert video1.channel == channel
|
||||
assert not video2.channel
|
||||
# Serializers do not query: load what Video's `__json__` reads, as the queries that serialize Videos do.
|
||||
video1 = test_session.query(Video).options(*Video.json_options()).populate_existing() \
|
||||
.filter(Video.id == video1.id).one()
|
||||
assert video1.__json__()['video']['channel']
|
||||
video_channel = video1.__json__()['video']['channel']
|
||||
assert video_channel['id'] == channel.id
|
||||
|
|
|
|||
|
|
@ -30,7 +30,7 @@ def get_video_for_app(session: Session, file_group_id: int, skip_viewed: bool =
|
|||
Get a Video by its FileGroup ID, with its prev/next videos. Mark the Video as viewed (a write; the
|
||||
caller commits).
|
||||
"""
|
||||
video = Video.find_by_file_group_id(session, file_group_id)
|
||||
video = Video.find_by_file_group_id(session, file_group_id, Video.json_options())
|
||||
if not skip_viewed:
|
||||
video.file_group.set_viewed()
|
||||
previous_video, next_video = video.get_surrounding_videos()
|
||||
|
|
|
|||
|
|
@ -323,7 +323,7 @@ def get_kiwix_catalog():
|
|||
|
||||
|
||||
def get_kiwix_subscriptions(session: Session) -> Dict[str, ZimSubscription]:
|
||||
subscriptions = session.query(ZimSubscription)
|
||||
subscriptions = session.query(ZimSubscription).options(*ZimSubscription.json_options())
|
||||
subscriptions_by_name = {i.name: i for i in subscriptions}
|
||||
return subscriptions_by_name
|
||||
|
||||
|
|
|
|||
|
|
@ -7,14 +7,14 @@ from typing import List, Tuple, OrderedDict as OrderedDictType, Dict, Optional,
|
|||
|
||||
from libzim import Archive, Searcher, Query, Entry, SuggestionSearcher
|
||||
from sqlalchemy import Column, Integer, BigInteger, ForeignKey, Text, tuple_, Boolean, UniqueConstraint
|
||||
from sqlalchemy.orm import relationship, Session
|
||||
from sqlalchemy.orm import relationship, Session, joinedload
|
||||
from sqlalchemy.orm.exc import NoResultFound # noqa
|
||||
|
||||
from modules.zim.errors import UnknownZimEntry, UnknownZimTagEntry, UnknownZim
|
||||
from wrolpi import dates, tags
|
||||
from wrolpi.common import Base, logger, get_relative_to_media_directory, ModelHelper
|
||||
from wrolpi.dates import TZDateTime
|
||||
from wrolpi.db import session_curs
|
||||
from wrolpi.db import session_curs, serializer
|
||||
from wrolpi.downloader import Download, download_manager
|
||||
from wrolpi.files.models import FileGroup
|
||||
from wrolpi.media_path import MediaPathType
|
||||
|
|
@ -80,6 +80,12 @@ class Zim(Base, ModelHelper):
|
|||
def __repr__(self):
|
||||
return f'<Zim id={self.id} file_group_id={self.file_group_id} path={self.path}>'
|
||||
|
||||
@staticmethod
|
||||
def json_options() -> tuple:
|
||||
"""Loader options for a query whose Zims will be serialized: `__json__` reads the FileGroup's size."""
|
||||
return joinedload(Zim.file_group),
|
||||
|
||||
@serializer
|
||||
def __json__(self) -> dict:
|
||||
d = dict(
|
||||
id=self.id,
|
||||
|
|
@ -319,7 +325,7 @@ class Zims:
|
|||
def get_all(session: Session) -> List[Zim]:
|
||||
"""Returns all Zim records."""
|
||||
zims = list()
|
||||
for zim in session.query(Zim).order_by(Zim.path):
|
||||
for zim in session.query(Zim).options(*Zim.json_options()).order_by(Zim.path):
|
||||
try:
|
||||
# It may not be possible to read the Zim file (drive not mounted), report and ignore.
|
||||
zim.get_zim()
|
||||
|
|
@ -390,6 +396,12 @@ class ZimSubscription(Base):
|
|||
download_id = self.download_id
|
||||
return f'<ZimSubscription id={self.id} {name=} {language=} {download_id=}>'
|
||||
|
||||
@staticmethod
|
||||
def json_options() -> tuple:
|
||||
"""Loader options for a query whose subscriptions will be serialized: `__json__` reads the Download's URL."""
|
||||
return joinedload(ZimSubscription.download).defer(Download.info_json),
|
||||
|
||||
@serializer
|
||||
def __json__(self) -> dict:
|
||||
d = dict(
|
||||
id=self.id,
|
||||
|
|
|
|||
|
|
@ -4,6 +4,8 @@ testpaths =
|
|||
wrolpi
|
||||
modules
|
||||
asyncio_default_fixture_loop_scope = function
|
||||
markers =
|
||||
allow_unexpected_queries: the test trips the serializer query guard on purpose (see wrolpi.db.no_db_access)
|
||||
# Do not remove timeout - tests should not run forever.
|
||||
#
|
||||
# This is a hang detector, not a performance budget. It was 30s, but the plugin was never
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ from pathlib import Path
|
|||
from sanic import response, HTTPResponse, Request, Sanic, SanicException
|
||||
|
||||
from wrolpi.common import Base, get_media_directory, logger, LOGGING_CONFIG
|
||||
from wrolpi.db import RequestDB
|
||||
from wrolpi.db import RequestDB, no_db_access
|
||||
from wrolpi.errors import APIError
|
||||
from wrolpi.perpetual import PERPETUAL_LOOPS, PER_WORKER_TASKS, PerpetualLoop, run_loop
|
||||
from wrolpi.vars import PYTEST
|
||||
|
|
@ -46,7 +46,8 @@ def json_response(*a, **kwargs) -> HTTPResponse:
|
|||
"""
|
||||
Handles encoding date/datetime in JSON.
|
||||
"""
|
||||
resp = response.json(*a, **kwargs, cls=CustomJSONEncoder, dumps=json.dumps)
|
||||
with no_db_access('json_response'):
|
||||
resp = response.json(*a, **kwargs, cls=CustomJSONEncoder, dumps=json.dumps)
|
||||
return resp
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ from wrolpi.errors import ValidationError
|
|||
from wrolpi.schema import JSONErrorResponse
|
||||
from . import lib, schema
|
||||
from .errors import UnknownCollection
|
||||
from .models import CollectionItem
|
||||
|
||||
# Create blueprint
|
||||
collection_bp = Blueprint('Collection', url_prefix='/api/collections')
|
||||
|
|
@ -98,6 +99,9 @@ async def add_collection_item_endpoint(request: Request, collection_id: int, bod
|
|||
url=body.url, title=body.title, position=body.position)
|
||||
# 201 when a new item was created; 200 when the item already existed (idempotent add).
|
||||
status = HTTPStatus.CREATED if created else HTTPStatus.OK
|
||||
# Load what `dict()` serializes (the FileGroup's Tags) before serializing.
|
||||
item = session.query(CollectionItem).options(*CollectionItem.json_options()).populate_existing() \
|
||||
.filter(CollectionItem.id == item.id).one()
|
||||
return json_response({'item': item.dict()}, status=status)
|
||||
except UnknownCollection as e:
|
||||
return json_response({'error': str(e)}, status=HTTPStatus.NOT_FOUND)
|
||||
|
|
|
|||
|
|
@ -14,7 +14,7 @@ from wrolpi import flags
|
|||
from wrolpi.common import logger, get_relative_to_media_directory, TRACE_LEVEL, background_task
|
||||
from wrolpi.db import get_db_session
|
||||
from wrolpi.errors import FileWorkerConflict
|
||||
from wrolpi.errors import ValidationError
|
||||
from wrolpi.errors import UnknownCollection, ValidationError
|
||||
from wrolpi.events import Events
|
||||
from wrolpi.tags import Tag
|
||||
from .models import Collection, validate_collection_directory
|
||||
|
|
@ -183,7 +183,9 @@ def get_collection_with_stats(session: Session, collection_id: int) -> dict:
|
|||
Raises:
|
||||
UnknownCollection: If collection not found
|
||||
"""
|
||||
collection = Collection.find_by_id(session, collection_id)
|
||||
collection = session.query(Collection).options(*Collection.json_options()).filter_by(id=collection_id).one_or_none()
|
||||
if not collection:
|
||||
raise UnknownCollection(f'Cannot find Collection with id {collection_id}')
|
||||
|
||||
# Get base collection data
|
||||
data = collection.__json__()
|
||||
|
|
@ -236,7 +238,9 @@ def get_collection_with_stats(session: Session, collection_id: int) -> dict:
|
|||
|
||||
# Playlists are manually curated and ordered, so return the full ordered item list.
|
||||
if collection.kind == 'playlist':
|
||||
data['items'] = [item.dict() for item in collection.items]
|
||||
items = session.query(CollectionItem).options(*CollectionItem.json_options()) \
|
||||
.filter(CollectionItem.collection_id == collection_id).order_by(CollectionItem.position)
|
||||
data['items'] = [item.dict() for item in items]
|
||||
|
||||
return data
|
||||
|
||||
|
|
|
|||
|
|
@ -3,12 +3,13 @@ from typing import Optional, List
|
|||
|
||||
from sqlalchemy import Column, Integer, String, ForeignKey, Text, DateTime, Index, UniqueConstraint, func, BigInteger, \
|
||||
or_, CheckConstraint
|
||||
from sqlalchemy.orm import relationship, Session
|
||||
from sqlalchemy.orm import relationship, Session, joinedload, selectinload
|
||||
from sqlalchemy.orm.collections import InstrumentedList
|
||||
|
||||
from wrolpi import flags
|
||||
from wrolpi.common import Base, ModelHelper, logger, get_media_directory, get_relative_to_media_directory, \
|
||||
unique_by_predicate, TRACE_LEVEL
|
||||
from wrolpi.db import serializer
|
||||
from wrolpi.downloader import Download, save_downloads_config
|
||||
from wrolpi.errors import ValidationError
|
||||
from wrolpi.events import Events
|
||||
|
|
@ -89,7 +90,8 @@ class Collection(ModelHelper, Base):
|
|||
|
||||
# Optional tag relationship (similar to Channel)
|
||||
tag_id = Column(Integer, ForeignKey('tag.id'))
|
||||
tag = relationship('Tag', primaryjoin='Collection.tag_id==Tag.id')
|
||||
# Joined: one narrow row, read by the Collection's (and its Channel's) `tag_name`.
|
||||
tag = relationship('Tag', primaryjoin='Collection.tag_id==Tag.id', lazy='joined')
|
||||
|
||||
# Stores the file_name_format used when files were last organized
|
||||
# Used to detect when reorganization is needed after config changes
|
||||
|
|
@ -114,7 +116,7 @@ class Collection(ModelHelper, Base):
|
|||
downloads: InstrumentedList = relationship(
|
||||
'Download',
|
||||
primaryjoin='Download.collection_id==Collection.id',
|
||||
back_populates='collection'
|
||||
back_populates='collection',
|
||||
)
|
||||
|
||||
def __repr__(self):
|
||||
|
|
@ -943,6 +945,13 @@ class Collection(ModelHelper, Base):
|
|||
self.tag_id = self.tag.id if self.tag else None
|
||||
return self.tag
|
||||
|
||||
@staticmethod
|
||||
def json_options() -> tuple:
|
||||
"""Loader options for a query whose Collections will be serialized: `__json__` lists the Downloads
|
||||
(without their yt-dlp info_json, which is not serialized)."""
|
||||
return selectinload(Collection.downloads).defer(Download.info_json),
|
||||
|
||||
@serializer
|
||||
def __json__(self) -> dict:
|
||||
"""Return JSON-serializable dict for API responses.
|
||||
|
||||
|
|
@ -1213,6 +1222,12 @@ class CollectionItem(ModelHelper, Base):
|
|||
return (f'<CollectionItem collection={self.collection_id} kind={self.item_kind} '
|
||||
f'position={self.position}>')
|
||||
|
||||
@staticmethod
|
||||
def json_options() -> tuple:
|
||||
"""Loader options for a query whose items will be serialized (`dict()`): each reads its FileGroup and the
|
||||
FileGroup's Tags, or its Zim."""
|
||||
return joinedload(CollectionItem.file_group).selectinload(FileGroup.tag_files), joinedload(CollectionItem.zim)
|
||||
|
||||
def dict(self) -> dict:
|
||||
"""Return dictionary representation."""
|
||||
d = super(CollectionItem, self).dict()
|
||||
|
|
|
|||
|
|
@ -145,16 +145,18 @@ async def test_add_file_item(async_client, test_session, test_directory):
|
|||
path.write_bytes(b'%PDF-1.4 test')
|
||||
fg = FileGroup.from_paths(test_session, path)
|
||||
test_session.commit()
|
||||
# Each request gets its own Session, as in production: requests commit, and in the shared test Session
|
||||
# that would expire the FileGroup the next request serializes.
|
||||
with production_like_sessions(test_session):
|
||||
_, response = await async_client.post(f'/api/collections/{cid}/items', json={
|
||||
'item_kind': 'file', 'file_group_id': fg.id})
|
||||
assert response.status_code == HTTPStatus.CREATED, response.json
|
||||
assert response.json['item']['item_kind'] == 'file'
|
||||
|
||||
_, response = await async_client.post(f'/api/collections/{cid}/items', json={
|
||||
'item_kind': 'file', 'file_group_id': fg.id})
|
||||
assert response.status_code == HTTPStatus.CREATED, response.json
|
||||
assert response.json['item']['item_kind'] == 'file'
|
||||
|
||||
_, response = await async_client.get(f'/api/collections/{cid}')
|
||||
items = response.json['collection']['items']
|
||||
assert len(items) == 1
|
||||
assert items[0]['file_group']['id'] == fg.id
|
||||
_, response = await async_client.get(f'/api/collections/{cid}')
|
||||
items = response.json['collection']['items']
|
||||
assert len(items) == 1
|
||||
assert items[0]['file_group']['id'] == fg.id
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
@ -351,34 +353,37 @@ async def test_playlist_directory_sync(async_client, test_session, test_director
|
|||
path.write_bytes(b'%PDF-1.4 test')
|
||||
fg = FileGroup.from_paths(test_session, path)
|
||||
test_session.commit()
|
||||
_, rf = await async_client.post(f'/api/collections/{cid}/items',
|
||||
json={'item_kind': 'file', 'file_group_id': fg.id})
|
||||
file_item_id = rf.json['item']['id']
|
||||
# A url item.
|
||||
_, ru = await async_client.post(f'/api/collections/{cid}/items',
|
||||
json={'item_kind': 'url', 'url': '/map?lat=1&lon=2&z=3', 'title': 'Map Spot'})
|
||||
url_item_id = ru.json['item']['id']
|
||||
# Each request gets its own Session, as in production: requests commit, and in the shared test Session
|
||||
# that would expire the FileGroup the next request serializes.
|
||||
with production_like_sessions(test_session):
|
||||
_, rf = await async_client.post(f'/api/collections/{cid}/items',
|
||||
json={'item_kind': 'file', 'file_group_id': fg.id})
|
||||
file_item_id = rf.json['item']['id']
|
||||
# A url item.
|
||||
_, ru = await async_client.post(f'/api/collections/{cid}/items',
|
||||
json={'item_kind': 'url', 'url': '/map?lat=1&lon=2&z=3', 'title': 'Map Spot'})
|
||||
url_item_id = ru.json['item']['id']
|
||||
|
||||
playlist_dir = test_directory / 'playlists' / 'Sync Test'
|
||||
names = sorted(p.name for p in playlist_dir.iterdir())
|
||||
assert any(n.startswith('0001_') and n.endswith('lesson.pdf') for n in names), names
|
||||
assert any(n.startswith('0002_') and n.endswith('.html') for n in names), names
|
||||
# The file is a hard link to the source (same inode).
|
||||
link = next(p for p in playlist_dir.iterdir() if p.name.startswith('0001_'))
|
||||
assert link.stat().st_ino == path.stat().st_ino
|
||||
playlist_dir = test_directory / 'playlists' / 'Sync Test'
|
||||
names = sorted(p.name for p in playlist_dir.iterdir())
|
||||
assert any(n.startswith('0001_') and n.endswith('lesson.pdf') for n in names), names
|
||||
assert any(n.startswith('0002_') and n.endswith('.html') for n in names), names
|
||||
# The file is a hard link to the source (same inode).
|
||||
link = next(p for p in playlist_dir.iterdir() if p.name.startswith('0001_'))
|
||||
assert link.stat().st_ino == path.stat().st_ino
|
||||
|
||||
# Reorder so the url is first; prefixes must follow the new positions.
|
||||
await async_client.put(f'/api/collections/{cid}/items/order',
|
||||
json={'item_ids': [url_item_id, file_item_id]})
|
||||
names = sorted(p.name for p in playlist_dir.iterdir())
|
||||
assert any(n.startswith('0001_') and n.endswith('.html') for n in names), names
|
||||
assert any(n.startswith('0002_') and n.endswith('lesson.pdf') for n in names), names
|
||||
# Reorder so the url is first; prefixes must follow the new positions.
|
||||
await async_client.put(f'/api/collections/{cid}/items/order',
|
||||
json={'item_ids': [url_item_id, file_item_id]})
|
||||
names = sorted(p.name for p in playlist_dir.iterdir())
|
||||
assert any(n.startswith('0001_') and n.endswith('.html') for n in names), names
|
||||
assert any(n.startswith('0002_') and n.endswith('lesson.pdf') for n in names), names
|
||||
|
||||
# Removing the url item drops its stub and re-sequences the file to 0001_.
|
||||
await async_client.delete(f'/api/collections/{cid}/items/{url_item_id}')
|
||||
names = sorted(p.name for p in playlist_dir.iterdir())
|
||||
assert not any(n.endswith('.html') for n in names), names
|
||||
assert any(n.startswith('0001_') and n.endswith('lesson.pdf') for n in names), names
|
||||
# Removing the url item drops its stub and re-sequences the file to 0001_.
|
||||
await async_client.delete(f'/api/collections/{cid}/items/{url_item_id}')
|
||||
names = sorted(p.name for p in playlist_dir.iterdir())
|
||||
assert not any(n.endswith('.html') for n in names), names
|
||||
assert any(n.startswith('0001_') and n.endswith('lesson.pdf') for n in names), names
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
|
|||
|
|
@ -100,6 +100,20 @@ def test_session(test_directory) -> Generator[Session, Any, None]:
|
|||
test_engine.dispose()
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def no_unexpected_queries(request):
|
||||
"""Fail any test during which a serializer ran SQL, even if the code under test caught the error."""
|
||||
from wrolpi import db
|
||||
db.UNEXPECTED_QUERIES.clear()
|
||||
yield
|
||||
if request.node.get_closest_marker('allow_unexpected_queries'):
|
||||
return
|
||||
queries = list(db.UNEXPECTED_QUERIES)
|
||||
db.UNEXPECTED_QUERIES.clear()
|
||||
assert not queries, 'A serializer ran SQL (a relationship or deferred column was not loaded):\n' + \
|
||||
'\n'.join(f'{reason}: {" ".join(statement.split())[:300]}' for reason, statement in queries)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def test_debug_logger(request):
|
||||
"""Tests default to DEBUG logging. Use pytest -vvv for TRACE level."""
|
||||
|
|
|
|||
58
wrolpi/db.py
58
wrolpi/db.py
|
|
@ -8,6 +8,7 @@ fresh session per use. Every connection gets the same PRAGMAs (WAL, busy_timeou
|
|||
via `create_wrolpi_engine` — use that factory for any engine touching a WROLPi database.
|
||||
"""
|
||||
import contextvars
|
||||
import functools
|
||||
import pathlib
|
||||
import sqlite3
|
||||
import threading
|
||||
|
|
@ -35,6 +36,61 @@ logger = logger.getChild(__name__)
|
|||
_immediate_txn = contextvars.ContextVar('wrolpi_immediate_txn', default=False)
|
||||
|
||||
|
||||
# The name of the serializer that is running (see `no_db_access`), or None.
|
||||
_no_db_access = contextvars.ContextVar('wrolpi_no_db_access', default=None)
|
||||
# Serializers that have already logged an unexpected query in this process (see `_refuse_query_while_serializing`).
|
||||
_warned_serializers = set()
|
||||
# Every unexpected query during tests, as (reason, statement). Recorded as well as raised, so code that catches
|
||||
# broad exceptions around a serializer cannot hide one; see the `no_unexpected_queries` fixture.
|
||||
UNEXPECTED_QUERIES = list()
|
||||
|
||||
|
||||
class UnexpectedQuery(RuntimeError):
|
||||
"""SQL ran inside `no_db_access` (while serializing)."""
|
||||
|
||||
|
||||
@contextmanager
|
||||
def no_db_access(reason: str):
|
||||
"""No SQL may run inside this block: the data it reads must already be loaded.
|
||||
|
||||
A query inside raises `UnexpectedQuery` during tests. In production it is logged once per `reason` and
|
||||
allowed, so a missed eager load costs a query, not a failed request."""
|
||||
token = _no_db_access.set(reason)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
_no_db_access.reset(token)
|
||||
|
||||
|
||||
def serializer(method):
|
||||
"""Run a model's serializer (`__json__`) inside `no_db_access`."""
|
||||
|
||||
@functools.wraps(method)
|
||||
def wrapper(self, *args, **kwargs):
|
||||
with no_db_access(f'{type(self).__name__}.{method.__name__}'):
|
||||
return method(self, *args, **kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
# Transaction control is not a query. The engine's `begin` listener emits BEGIN through this same event, as the
|
||||
# first statement on a Session with no open transaction; the query that follows is the one to report.
|
||||
_TRANSACTION_CONTROL = ('BEGIN', 'COMMIT', 'ROLLBACK', 'SAVEPOINT', 'RELEASE')
|
||||
|
||||
|
||||
def _refuse_query_while_serializing(conn, cursor, statement, parameters, context, executemany):
|
||||
reason = _no_db_access.get()
|
||||
if reason is None or statement.lstrip().upper().startswith(_TRANSACTION_CONTROL):
|
||||
return
|
||||
if PYTEST:
|
||||
UNEXPECTED_QUERIES.append((reason, statement))
|
||||
raise UnexpectedQuery(f'{reason} ran SQL while serializing: {statement}')
|
||||
if reason not in _warned_serializers:
|
||||
_warned_serializers.add(reason)
|
||||
logger.warning(f'{reason} ran SQL while serializing: a relationship or deferred column it reads was not '
|
||||
f'loaded, or a commit expired the object: {statement}')
|
||||
|
||||
|
||||
def _adapt_datetime(value):
|
||||
"""Store datetimes from raw SQL exactly like SQLAlchemy does: naive UTC with microseconds."""
|
||||
from datetime import timezone
|
||||
|
|
@ -131,6 +187,8 @@ def create_wrolpi_engine(target: Union[str, pathlib.Path]) -> sqlalchemy.engine.
|
|||
def _sqlite_on_connect(dbapi_conn, _):
|
||||
journal_state['mode'] = _configure_sqlite_connection(dbapi_conn, journal_state['mode'])
|
||||
|
||||
event.listen(engine, 'before_cursor_execute', _refuse_query_while_serializing)
|
||||
|
||||
@event.listens_for(engine, 'begin')
|
||||
def _sqlite_do_begin(conn):
|
||||
# Write sessions (`commit=True`) take the write lock up front; read sessions stay deferred
|
||||
|
|
|
|||
|
|
@ -35,7 +35,7 @@ from wrolpi.common import Base, ModelHelper, logger, wrol_mode_check, zig_zag, C
|
|||
wrol_mode_enabled, background_task, get_absolute_media_path, timer, aiohttp_get, \
|
||||
get_download_info, trim_file_name, get_wrolpi_config, TRACE_LEVEL, normalize_domain
|
||||
from wrolpi.dates import TZDateTime, now, Seconds
|
||||
from wrolpi.db import get_db_session, session_curs
|
||||
from wrolpi.db import get_db_session, session_curs, serializer
|
||||
from wrolpi.errors import InvalidDownload, UnrecoverableDownloadError, BotBlockedDownloadError, UnknownDownload, \
|
||||
ValidationError, DownloadError
|
||||
from wrolpi.events import Events
|
||||
|
|
@ -280,6 +280,7 @@ class Download(ModelHelper, Base): # noqa
|
|||
return f'<Download id={self.id} status={self.status} url={repr(self.url)} attempts={self.attempts} ' \
|
||||
f'error={bool(self.error)}>'
|
||||
|
||||
@serializer
|
||||
def __json__(self) -> dict:
|
||||
d = dict(
|
||||
attempts=self.attempts,
|
||||
|
|
|
|||
|
|
@ -22,7 +22,7 @@ from typing import Callable, List, Tuple, Union, Dict, Generator, Iterable, Set
|
|||
import cachetools.func
|
||||
from sqlalchemy import asc, or_, text as sa_text
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy.orm import Session, undefer, joinedload
|
||||
|
||||
from wrolpi.cmd import which
|
||||
from wrolpi.common import get_media_directory, wrol_mode_check, logger, \
|
||||
|
|
@ -234,6 +234,7 @@ def get_tagged_file_groups_by_ids(session: Session, file_group_ids: List[int]) -
|
|||
if not file_group_ids:
|
||||
return []
|
||||
rows = session.query(FileGroup) \
|
||||
.options(*FileGroup.json_options()) \
|
||||
.join(TagFile, TagFile.file_group_id == FileGroup.id) \
|
||||
.filter(FileGroup.id.in_(file_group_ids)) \
|
||||
.distinct() \
|
||||
|
|
@ -255,6 +256,7 @@ def get_tagged_file_groups_by_paths(session: Session, paths: Iterable[pathlib.Pa
|
|||
seen_ids = set()
|
||||
for path in paths:
|
||||
query = session.query(FileGroup) \
|
||||
.options(*FileGroup.json_options()) \
|
||||
.join(TagFile, TagFile.file_group_id == FileGroup.id)
|
||||
if path.is_file():
|
||||
# Could be deleting a file in a FileGroup that has been tagged.
|
||||
|
|
@ -1330,7 +1332,10 @@ def handle_file_group_search_results(session: Session, statement: str, params: d
|
|||
|
||||
from modules.archive.models import Archive
|
||||
from modules.videos.models import Video
|
||||
# Serialized below: what Video's and Archive's `__json__` read beyond the FileGroup (see their `json_options`).
|
||||
results = session.query(FileGroup, Video, Archive) \
|
||||
.options(*FileGroup.json_options(), undefer(Video.ffprobe_json), joinedload(Video.channel),
|
||||
joinedload(Archive.collection)) \
|
||||
.filter(FileGroup.id.in_(ordered_ids)) \
|
||||
.outerjoin(Video, Video.file_group_id == FileGroup.id) \
|
||||
.outerjoin(Archive, Archive.file_group_id == FileGroup.id)
|
||||
|
|
|
|||
|
|
@ -8,12 +8,12 @@ from typing import List, Type, Optional, Iterable
|
|||
|
||||
from sqlalchemy import Column, String, BigInteger, Boolean, event, Index, Integer, JSON, or_
|
||||
from sqlalchemy import types
|
||||
from sqlalchemy.orm import deferred, relationship, Session
|
||||
from sqlalchemy.orm import deferred, relationship, Session, selectinload
|
||||
|
||||
from wrolpi.common import Base, ModelHelper, logger, recursive_map, get_media_directory, \
|
||||
get_relative_to_media_directory, unique_by_predicate, replace_file
|
||||
from wrolpi.dates import TZDateTime, now, from_timestamp, strptime_ms, strftime
|
||||
from wrolpi.db import get_db_session
|
||||
from wrolpi.db import get_db_session, serializer
|
||||
from wrolpi.downloader import Download
|
||||
from wrolpi.errors import FileGroupIsTagged, UnknownFile
|
||||
from wrolpi.files import indexers
|
||||
|
|
@ -134,6 +134,12 @@ class FileGroup(ModelHelper, Base):
|
|||
m = f'model={self.model}' if self.model else f'mimetype={self.mimetype}'
|
||||
return f'<FileGroup id={self.id} {m} primary_path={repr(str(self.primary_path))}>'
|
||||
|
||||
@staticmethod
|
||||
def json_options() -> tuple:
|
||||
"""Loader options for a query whose FileGroups will be serialized: `__json__` reads their Tags."""
|
||||
return selectinload(FileGroup.tag_files),
|
||||
|
||||
@serializer
|
||||
def __json__(self) -> dict:
|
||||
from wrolpi.files.lib import split_path_stem_and_suffix
|
||||
_, suffix = split_path_stem_and_suffix(self.primary_path)
|
||||
|
|
@ -854,6 +860,7 @@ class Directory(ModelHelper, Base):
|
|||
name: str = Column(String, nullable=False)
|
||||
idempotency = Column(TZDateTime, default=lambda: now())
|
||||
|
||||
@serializer
|
||||
def __json__(self) -> dict:
|
||||
d = dict(
|
||||
path=self.path,
|
||||
|
|
|
|||
|
|
@ -14,7 +14,7 @@ from wrolpi import dates, flags
|
|||
from wrolpi.common import ModelHelper, Base, logger, ConfigFile, get_media_directory, background_task, \
|
||||
get_relative_to_media_directory, is_valid_hex_color, walk, INVALID_FILE_CHARS, get_wrolpi_config
|
||||
from wrolpi.dates import TZDateTime
|
||||
from wrolpi.db import get_db_session, named_placeholders, session_curs
|
||||
from wrolpi.db import get_db_session, named_placeholders, session_curs, serializer
|
||||
from wrolpi.downloader import save_downloads_config
|
||||
from wrolpi.errors import UnknownTag, UsedTag, InvalidTag, FileWorkerConflict, NoPrimaryFile
|
||||
from wrolpi.events import Events
|
||||
|
|
@ -36,7 +36,8 @@ class TagFile(ModelHelper, Base):
|
|||
created_at: datetime = Column(TZDateTime, default=dates.now)
|
||||
|
||||
tag_id = Column(Integer, ForeignKey('tag.id'), primary_key=True)
|
||||
tag = relationship('Tag', back_populates='tag_files')
|
||||
# Joined: a TagFile is loaded for its Tag's name (serializers, the tags config).
|
||||
tag = relationship('Tag', back_populates='tag_files', lazy='joined')
|
||||
file_group_id = Column(BigInteger, ForeignKey('file_group.id', ondelete='CASCADE'), primary_key=True)
|
||||
file_group = relationship('FileGroup', back_populates='tag_files')
|
||||
|
||||
|
|
@ -86,6 +87,7 @@ class Tag(ModelHelper, Base):
|
|||
color = self.color
|
||||
return f'<Tag {name=} {color=}>'
|
||||
|
||||
@serializer
|
||||
def __json__(self) -> dict:
|
||||
return dict(
|
||||
id=self.id,
|
||||
|
|
|
|||
|
|
@ -1,10 +1,13 @@
|
|||
import asyncio
|
||||
import logging
|
||||
import sqlite3
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
from wrolpi.conftest import production_like_sessions, probe_write_lock_is_held, count_connections
|
||||
from wrolpi.db import get_db_session, _configure_sqlite_connection, RequestDB, RequestSession
|
||||
from wrolpi.db import get_db_session, _configure_sqlite_connection, RequestDB, RequestSession, no_db_access, \
|
||||
UnexpectedQuery, serializer
|
||||
from wrolpi.tags import Tag
|
||||
|
||||
|
||||
|
|
@ -256,3 +259,65 @@ def test_request_session_keeps_one_connection_until_closed(test_session):
|
|||
session.query(Tag).count()
|
||||
assert len(opened) == 2, 'a closed RequestSession did not reconnect'
|
||||
session.close()
|
||||
|
||||
|
||||
@pytest.mark.allow_unexpected_queries
|
||||
def test_no_db_access_refuses_queries_during_tests(test_session):
|
||||
"""Serializing must not query: the data must already be loaded."""
|
||||
test_session.query(Tag).count() # Allowed outside the block.
|
||||
with pytest.raises(UnexpectedQuery, match='a test serializer ran SQL'):
|
||||
with no_db_access('a test serializer'):
|
||||
test_session.query(Tag).count()
|
||||
test_session.rollback()
|
||||
assert test_session.query(Tag).count() == 0, 'queries work again after the block'
|
||||
|
||||
|
||||
@pytest.mark.allow_unexpected_queries
|
||||
def test_serializer_decorator_guards_the_method(test_session):
|
||||
class Thing:
|
||||
@serializer
|
||||
def __json__(self):
|
||||
return dict(count=test_session.query(Tag).count())
|
||||
|
||||
with pytest.raises(UnexpectedQuery, match=r'Thing\.__json__'):
|
||||
Thing().__json__()
|
||||
|
||||
|
||||
def test_no_db_access_logs_once_and_allows_in_production(test_session, caplog):
|
||||
"""In production a missed eager load costs a query, not a failed request; it is logged once."""
|
||||
with mock.patch('wrolpi.db.PYTEST', False), mock.patch('wrolpi.db._warned_serializers', set()), \
|
||||
caplog.at_level(logging.WARNING, logger='wrolpi.db'):
|
||||
with no_db_access('a production serializer'):
|
||||
assert test_session.query(Tag).count() == 0
|
||||
assert test_session.query(Tag).count() == 0
|
||||
warnings = [r for r in caplog.records if 'a production serializer ran SQL' in r.getMessage()]
|
||||
assert len(warnings) == 1
|
||||
|
||||
|
||||
@pytest.mark.allow_unexpected_queries
|
||||
def test_no_db_access_reports_the_query_not_the_begin(test_session):
|
||||
"""On a Session with no open transaction the engine emits BEGIN first; the guard names the real query."""
|
||||
from wrolpi import db
|
||||
test_session.commit()
|
||||
with production_like_sessions(test_session) as maker:
|
||||
session = maker()
|
||||
with pytest.raises(UnexpectedQuery, match='SELECT'):
|
||||
with no_db_access('a serializer on a fresh session'):
|
||||
session.query(Tag).count()
|
||||
session.close()
|
||||
assert [statement.split()[0] for _, statement in db.UNEXPECTED_QUERIES] == ['SELECT']
|
||||
db.UNEXPECTED_QUERIES.clear()
|
||||
|
||||
|
||||
def test_no_db_access_logs_the_query_in_production(test_session, caplog):
|
||||
"""The single production warning per serializer names the query that ran, not the transaction's BEGIN."""
|
||||
test_session.commit()
|
||||
with production_like_sessions(test_session) as maker, mock.patch('wrolpi.db.PYTEST', False), \
|
||||
mock.patch('wrolpi.db._warned_serializers', set()), caplog.at_level(logging.WARNING, logger='wrolpi.db'):
|
||||
session = maker()
|
||||
with no_db_access('a production serializer'):
|
||||
session.query(Tag).count()
|
||||
session.close()
|
||||
warnings = [r.getMessage() for r in caplog.records if 'a production serializer ran SQL' in r.getMessage()]
|
||||
assert len(warnings) == 1
|
||||
assert 'SELECT' in warnings[0] and 'BEGIN' not in warnings[0]
|
||||
|
|
|
|||
63
wrolpi/test/test_serializers.py
Normal file
63
wrolpi/test/test_serializers.py
Normal file
|
|
@ -0,0 +1,63 @@
|
|||
"""Serializers read only data that is already loaded (see `wrolpi.db.no_db_access`).
|
||||
|
||||
A lazy load while serializing is a hidden query per row. Narrow many-to-one relationships load with their parent
|
||||
(`lazy='joined'`); a query whose results are serialized adds the model's `json_options()` for the rest.
|
||||
"""
|
||||
import pytest
|
||||
|
||||
from modules.archive.models import Archive
|
||||
from modules.docs.models import Doc
|
||||
from modules.videos.models import Channel, Video
|
||||
from modules.zim.models import Zim, ZimSubscription
|
||||
from wrolpi.collections.models import Collection, CollectionItem
|
||||
from wrolpi.conftest import production_like_sessions
|
||||
from wrolpi.db import no_db_access
|
||||
from wrolpi.downloader import Download
|
||||
from wrolpi.files.models import FileGroup
|
||||
from wrolpi.tags import Tag
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_models_serialize_without_queries(async_client, test_session, tag_factory, simple_channel,
|
||||
video_factory, archive_factory, zim_factory, test_downloader,
|
||||
test_directory):
|
||||
tag = await tag_factory()
|
||||
video = video_factory(title='a video', channel_id=simple_channel.id)
|
||||
video.file_group.add_tag(test_session, tag.name)
|
||||
archive_factory(domain='example.com')
|
||||
doc_path = test_directory / 'manual.pdf'
|
||||
doc_path.write_bytes(b'%PDF-1.4 test')
|
||||
test_session.add(Doc(file_group=FileGroup.from_paths(test_session, doc_path)))
|
||||
zim_factory('wikipedia.zim')
|
||||
simple_channel.collection.tag = tag
|
||||
download = Download(url='https://example.com/channel', downloader=test_downloader.name, frequency=3600,
|
||||
collection_id=simple_channel.collection_id)
|
||||
test_session.add(download)
|
||||
test_session.flush()
|
||||
test_session.add(ZimSubscription(name='Wikipedia', language='en', download_id=download.id))
|
||||
test_session.commit()
|
||||
|
||||
_, response = await async_client.post('/api/collections', json={'name': 'Mixed'})
|
||||
playlist_id = response.json['collection']['id']
|
||||
for body in ({'item_kind': 'file', 'file_group_id': video.file_group_id},
|
||||
{'item_kind': 'url', 'url': '/videos', 'title': 'Videos'}):
|
||||
await async_client.post(f'/api/collections/{playlist_id}/items', json=body)
|
||||
|
||||
with production_like_sessions(test_session) as maker:
|
||||
session = maker()
|
||||
for model in (FileGroup, Video, Archive, Doc, Zim, ZimSubscription, Collection, Channel, Tag, Download):
|
||||
# Each serializing query loads what the model's serializer reads (`json_options`).
|
||||
query = session.query(model)
|
||||
if hasattr(model, 'json_options'):
|
||||
query = query.options(*model.json_options())
|
||||
objects = query.all()
|
||||
assert objects, f'no {model.__name__} to serialize'
|
||||
with no_db_access(f'serializing {model.__name__}'):
|
||||
for obj in objects:
|
||||
obj.__json__()
|
||||
|
||||
items = session.query(CollectionItem).options(*CollectionItem.json_options()).all()
|
||||
assert len(items) == 2
|
||||
with no_db_access('serializing CollectionItem'):
|
||||
serialized = [item.dict() for item in items]
|
||||
assert serialized[0]['file_group']['tags'] == [tag.name]
|
||||
|
|
@ -37,6 +37,8 @@ async def test_tags_file_group_json(async_client, test_session, make_files_struc
|
|||
async def test_tags_model(async_client, test_session, make_files_structure, tag_factory, example_pdf):
|
||||
"""Can get Tag using class methods."""
|
||||
tag1 = await tag_factory()
|
||||
# Serializers do not query; reload the Tag the factory's commit expired.
|
||||
test_session.refresh(tag1)
|
||||
assert tag1.__json__()
|
||||
|
||||
assert Tag.find_by_id(test_session, tag1.id)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue