mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 02:48:38 -04:00
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>
This commit is contained in:
Roland
parent
72a0379a7f
commit
021bdad947
18 changed files with 153 additions and 58 deletions
|
|
@ -5,7 +5,7 @@ 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
|
||||
|
|
@ -36,10 +36,15 @@ class Archive(Base, ModelHelper):
|
|||
id = Column(Integer, primary_key=True)
|
||||
|
||||
collection_id = Column(Integer, ForeignKey('collection.id', ondelete='CASCADE'))
|
||||
collection: Collection = relationship('Collection', primaryjoin='Archive.collection_id==Collection.id',
|
||||
lazy='joined')
|
||||
collection: Collection = relationship('Collection', primaryjoin='Archive.collection_id==Collection.id')
|
||||
file_group_id = Column(BigInteger, ForeignKey('file_group.id', ondelete='CASCADE'), unique=True, nullable=False)
|
||||
file_group: FileGroup = relationship('FileGroup', lazy='joined')
|
||||
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:
|
||||
|
|
|
|||
|
|
@ -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,7 +2,7 @@ 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, serializer
|
||||
|
|
@ -56,7 +56,7 @@ class Doc(ModelHelper, Base):
|
|||
description = Column(Text)
|
||||
|
||||
file_group_id = Column(BigInteger, ForeignKey('file_group.id', ondelete='CASCADE'), nullable=False, unique=True)
|
||||
file_group: FileGroup = relationship('FileGroup', lazy='joined')
|
||||
file_group: FileGroup = relationship('FileGroup')
|
||||
sections = relationship('DocSection', back_populates='doc',
|
||||
cascade='all, delete-orphan',
|
||||
passive_deletes=True)
|
||||
|
|
@ -65,6 +65,11 @@ 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__()
|
||||
|
|
|
|||
|
|
@ -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,7 +3,7 @@ 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, undefer
|
||||
from sqlalchemy.orm import relationship, Session, deferred, undefer, joinedload, selectinload
|
||||
from sqlalchemy.orm.collections import InstrumentedList
|
||||
|
||||
from modules.videos.errors import UnknownVideo, UnknownChannel
|
||||
|
|
@ -11,6 +11,7 @@ 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, 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
|
||||
|
|
@ -46,10 +47,9 @@ class Video(ModelHelper, Base):
|
|||
comments_failed = Column(Boolean, default=False) # see `get_missing_videos_comments`
|
||||
|
||||
channel_id = Column(Integer, ForeignKey('channel.id'))
|
||||
channel = relationship('Channel', primaryjoin='Video.channel_id==Channel.id', back_populates='videos',
|
||||
lazy='joined')
|
||||
channel = relationship('Channel', primaryjoin='Video.channel_id==Channel.id', back_populates='videos')
|
||||
file_group_id = Column(BigInteger, ForeignKey('file_group.id', ondelete='CASCADE'), unique=True, nullable=False)
|
||||
file_group: FileGroup = relationship('FileGroup', lazy='joined')
|
||||
file_group: FileGroup = relationship('FileGroup')
|
||||
|
||||
def __repr__(self):
|
||||
v = None
|
||||
|
|
@ -58,6 +58,13 @@ 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__()
|
||||
|
|
@ -235,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
|
||||
|
||||
|
|
@ -393,30 +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."""
|
||||
# `__json__` reads the (deferred) ffprobe_json for the codecs.
|
||||
video = session.query(Video).options(undefer(Video.ffprobe_json)).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"""
|
||||
# `__json__` reads the (deferred) ffprobe_json for the codecs.
|
||||
video = session.query(Video).options(undefer(Video.ffprobe_json)) \
|
||||
.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
|
||||
|
|
@ -625,6 +630,7 @@ class Channel(ModelHelper, Base):
|
|||
|
||||
videos: InstrumentedList = relationship('Video', primaryjoin='Channel.id==Video.channel_id')
|
||||
collection_id = Column(Integer, ForeignKey('collection.id', ondelete='CASCADE'))
|
||||
# Joined: a Channel's name, directory and Tag live on its Collection.
|
||||
collection = relationship('Collection', foreign_keys=[collection_id], lazy='joined')
|
||||
|
||||
@property
|
||||
|
|
@ -943,6 +949,12 @@ 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(
|
||||
|
|
|
|||
|
|
@ -127,8 +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 the deferred ffprobe_json first, as the queries that serialize Videos do.
|
||||
test_session.refresh(video1, ['ffprobe_json'])
|
||||
# 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,7 +7,7 @@ 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
|
||||
|
|
@ -74,12 +74,17 @@ class Zim(Base, ModelHelper):
|
|||
path: pathlib.Path = Column(MediaPathType, nullable=False)
|
||||
|
||||
file_group_id = Column(BigInteger, ForeignKey('file_group.id', ondelete='CASCADE'), nullable=False)
|
||||
file_group: FileGroup = relationship('FileGroup', lazy='joined')
|
||||
file_group: FileGroup = relationship('FileGroup')
|
||||
auto_search = Column(Boolean, default=True)
|
||||
|
||||
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(
|
||||
|
|
@ -320,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()
|
||||
|
|
@ -383,8 +388,7 @@ class ZimSubscription(Base):
|
|||
name: str = Column(Text, unique=True, nullable=False)
|
||||
language: str = Column(Text, nullable=False)
|
||||
download_id: int = Column(Integer, ForeignKey('download.id', ondelete='CASCADE'), nullable=False)
|
||||
download: Download = relationship('Download', primaryjoin='ZimSubscription.download_id==Download.id',
|
||||
lazy='joined')
|
||||
download: Download = relationship('Download', primaryjoin='ZimSubscription.download_id==Download.id')
|
||||
|
||||
def __repr__(self):
|
||||
name = self.name
|
||||
|
|
@ -392,6 +396,11 @@ 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(
|
||||
|
|
|
|||
|
|
@ -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,7 +3,7 @@ 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
|
||||
|
|
@ -90,6 +90,7 @@ class Collection(ModelHelper, Base):
|
|||
|
||||
# Optional tag relationship (similar to Channel)
|
||||
tag_id = Column(Integer, ForeignKey('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
|
||||
|
|
@ -116,7 +117,6 @@ class Collection(ModelHelper, Base):
|
|||
'Download',
|
||||
primaryjoin='Download.collection_id==Collection.id',
|
||||
back_populates='collection',
|
||||
lazy='selectin',
|
||||
)
|
||||
|
||||
def __repr__(self):
|
||||
|
|
@ -945,6 +945,12 @@ 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.
|
||||
|
|
@ -1188,9 +1194,8 @@ class CollectionItem(ModelHelper, Base):
|
|||
|
||||
# Relationships
|
||||
collection = relationship('Collection', back_populates='items')
|
||||
# Eager: `dict()` serializes the item's FileGroup or Zim for every item of a listing.
|
||||
file_group: FileGroup = relationship('FileGroup', lazy='joined')
|
||||
zim = relationship('Zim', lazy='joined')
|
||||
file_group: FileGroup = relationship('FileGroup')
|
||||
zim = relationship('Zim')
|
||||
|
||||
# Indexes for performance
|
||||
__table_args__ = (
|
||||
|
|
@ -1217,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()
|
||||
|
|
|
|||
10
wrolpi/db.py
10
wrolpi/db.py
|
|
@ -36,8 +36,7 @@ logger = logger.getChild(__name__)
|
|||
_immediate_txn = contextvars.ContextVar('wrolpi_immediate_txn', default=False)
|
||||
|
||||
|
||||
# Set while serializing (`no_db_access`). A query then is a lazy load or an expired-attribute refresh: hidden
|
||||
# per-row SQL today, and a `MissingGreenlet` error under an async Session. Holds the name of what is serializing.
|
||||
# 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()
|
||||
|
|
@ -74,9 +73,14 @@ def serializer(method):
|
|||
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:
|
||||
if reason is None or statement.lstrip().upper().startswith(_TRANSACTION_CONTROL):
|
||||
return
|
||||
if PYTEST:
|
||||
UNEXPECTED_QUERIES.append((reason, statement))
|
||||
|
|
|
|||
|
|
@ -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, undefer
|
||||
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,9 +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
|
||||
# Video's `__json__` reads its (deferred) ffprobe_json for the codecs.
|
||||
# Serialized below: what Video's and Archive's `__json__` read beyond the FileGroup (see their `json_options`).
|
||||
results = session.query(FileGroup, Video, Archive) \
|
||||
.options(undefer(Video.ffprobe_json)) \
|
||||
.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,7 +8,7 @@ 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
|
||||
|
|
@ -121,8 +121,7 @@ class FileGroup(ModelHelper, Base):
|
|||
# `file_group_effective_datetime_trigger` and `update_effective_datetime` event handler
|
||||
effective_datetime = Column(TZDateTime) # Equivalent to COALESCE(published_datetime, download_datetime)
|
||||
|
||||
# Eager: `__json__` lists the Tags of every FileGroup it serializes.
|
||||
tag_files: Iterable[TagFile] = relationship('TagFile', cascade='all', lazy='selectin')
|
||||
tag_files: Iterable[TagFile] = relationship('TagFile', cascade='all')
|
||||
|
||||
# Full-text search over these columns is provided by the external-content FTS5 table
|
||||
# `file_group_fts` (see `wrolpi.fts`): a=title, b=author-ish, c=description, d=body/captions.
|
||||
|
|
@ -135,6 +134,11 @@ 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
|
||||
|
|
|
|||
|
|
@ -36,6 +36,7 @@ class TagFile(ModelHelper, Base):
|
|||
created_at: datetime = Column(TZDateTime, default=dates.now)
|
||||
|
||||
tag_id = Column(Integer, ForeignKey('tag.id'), primary_key=True)
|
||||
# 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')
|
||||
|
|
|
|||
|
|
@ -292,3 +292,32 @@ def test_no_db_access_logs_once_and_allows_in_production(test_session, caplog):
|
|||
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]
|
||||
|
|
|
|||
|
|
@ -1,10 +1,9 @@
|
|||
"""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, and a `MissingGreenlet` error under an async Session.
|
||||
The relationships each serializer reads are loaded with their parent (`lazy='joined'` / `'selectin'`).
|
||||
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 sqlalchemy.orm import undefer
|
||||
|
||||
from modules.archive.models import Archive
|
||||
from modules.docs.models import Doc
|
||||
|
|
@ -47,17 +46,17 @@ async def test_models_serialize_without_queries(async_client, test_session, tag_
|
|||
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 model is Video:
|
||||
# Like the queries that serialize Videos: the codecs come from the deferred ffprobe_json.
|
||||
query = query.options(undefer(Video.ffprobe_json))
|
||||
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).all()
|
||||
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]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue