rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-25 18:40:06 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

db: request-path library functions use the request's connection

Browse source 
Library functions an API request reaches now take the request's Session (first argument) instead of
opening their own session or cursor, so a request needs one SQLite connection.  Raw SQL runs through the
new `wrolpi.db.session_curs(session)`, a cursor on the Session's connection; `RequestDB.curs()` uses it.

- Search: files, videos, archives, flasher, and docs search share the request connection for the total,
  the page of ids, and the page's models (3 connections -> 1).
- Zim search, including tag-filtered search and estimates (2 -> 1).
- Channel detail (`Channel.get_statistics` reads on the Channel's own Session) and the channel list.
- Video detail, edit, description, captions, comments and transcode.  `get_video_for_app` and
  `update_video` no longer commit; their handlers use `request.ctx.db.write()`, so editing a Video and
  reading it back is one transaction.
- Tags (list, overlaps, recent), file statistics and estimates, other estimates, bulk tag preview, the
  download list, retry, kill and the status summary, and docs/videos/archive statistics.

Not converted: file operations that refresh files inside the request (delete, rename, upload, channel
and collection tag/move) commit page by page in code shared with the file worker; the cached zim entry
lookup; config import.

The connection-count tests now assert one connection for every endpoint they cover (no xfails left), and
cover the videos and archive statistics endpoints, which had no API test.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Roland 2026-10-04 10:10:38 -06:00
parent a2bef48c6c
commit 24c5796fc6
No known key found for this signature in database
GPG key ID: 1BA620CE6C9A4680
34 changed files with 545 additions and 422 deletions
Download patch file Download diff file Expand all files Collapse all files

8
modules/archive/api.py
View file

@ -27,8 +27,8 @@ logger = logger.getChild(__name__)
@archive_bp.get('/statistics')
@openapi.response(HTTPStatus.OK, schema.ArchiveStatisticsResponse)
async def statistics(_: Request):
ret = await lib.get_statistics()
async def statistics(request: Request):
ret = await lib.get_statistics(request.ctx.session)
return json_response(ret, HTTPStatus.OK)
@ -55,14 +55,14 @@ archive_limit_limiter = api_param_limiter(100)
@openapi.response(HTTPStatus.OK, schema.ArchiveSearchResponse)
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
@validate(schema.ArchiveSearchRequest)
async def search_archives(_: Request, body: schema.ArchiveSearchRequest):
async def search_archives(request: Request, body: schema.ArchiveSearchRequest):
search_str = body.search_str
domain = body.domain
limit = archive_limit_limiter(body.limit)
offset = body.offset or 0
file_groups, total = await asyncio.to_thread(
lib.search_archives, search_str, domain, limit, offset, body.order_by, body.tag_names,
lib.search_archives, request.ctx.session, search_str, domain, limit, offset, body.order_by, body.tag_names,
body.headline, body.deep)
ret = dict(file_groups=file_groups, totals=dict(file_groups=total))
return json_response(ret)

16
modules/archive/lib.py
View file

@ -27,7 +27,7 @@ from wrolpi.common import get_media_directory, get_relative_to_media_directory,
escape_file_name, aiohttp_post, format_html_string, split_lines_by_length, get_html_soup, get_title_from_html, \
get_wrolpi_config, html_screenshot, html_file_screenshot, ConfigFile, trim_file_name
from wrolpi.dates import now, Seconds
from wrolpi.db import get_db_session, get_db_curs
from wrolpi.db import get_db_session, get_db_curs, session_curs
from wrolpi.errors import UnknownArchive, InvalidOrderBy, InvalidDatetime
from wrolpi.events import Events
from wrolpi.files.lib import handle_file_group_search_results, cached_search_total, \
@ -1390,8 +1390,8 @@ NO_NULL_ORDERS = {
}
def search_archives(search_str: str, domain: str, limit: int, offset: int, order: str, tag_names: List[str],
headline: bool = False, deep: bool = False) \
def search_archives(session: Session, search_str: str, domain: str, limit: int, offset: int, order: str,
tag_names: List[str], headline: bool = False, deep: bool = False) \
-> Tuple[List[dict], int]:
# Always filter FileGroups to Archives.
wheres = []
@ -1479,7 +1479,7 @@ def search_archives(search_str: str, domain: str, limit: int, offset: int, order
if unfiltered:
total = cached_search_total(
search_filter_cache_key('archives'),
lambda: count_file_groups('SELECT COUNT(*) AS total FROM archive', dict()),
lambda: count_file_groups(session, 'SELECT COUNT(*) AS total FROM archive', dict()),
)
else:
if file_group_search:
@ -1499,9 +1499,9 @@ def search_archives(search_str: str, domain: str, limit: int, offset: int, order
'archives', search_str=search_str, domain=domain, tag_names=tag_names,
deep=deep, null_filter=null_filter,
)
total = cached_search_total(cache_key, lambda: count_file_groups(count_stmt, params))
total = cached_search_total(cache_key, lambda: count_file_groups(session, count_stmt, params))
results, total = handle_file_group_search_results(stmt, params, total=total)
results, total = handle_file_group_search_results(session, stmt, params, total=total)
if file_group_search and headline and results:
# Highlight the plain titles like the FTS snippets above (Postgres used ts_headline).
@ -1657,9 +1657,9 @@ async def singlefile_to_archive(singlefile: bytes, destination: pathlib.Path = N
return archive
async def get_statistics():
async def get_statistics(session: Session):
"""Get statistics about Archives and Domain Collections."""
with get_db_curs() as curs:
with session_curs(session) as curs:
curs.execute('''
SELECT
-- total archives

6
modules/archive/test/test_lib.py
View file

@ -1071,6 +1071,7 @@ def test_search_archives_by_domain(test_directory, test_session, archive_factory
# Search for archives in searchtest.com
file_groups, total = search_archives(
test_session,
search_str=None,
domain='searchtest.com',
limit=10,
@ -1085,6 +1086,7 @@ def test_search_archives_by_domain(test_directory, test_session, archive_factory
# Search for other.com
file_groups, total = search_archives(
test_session,
search_str=None,
domain='other.com',
limit=10,
@ -1375,7 +1377,7 @@ async def test_get_statistics(test_session, archive_factory):
from modules.archive.lib import get_statistics
# Can get statistics in empty DB.
result = await get_statistics()
result = await get_statistics(test_session)
assert 'statistics' in result
assert 'archives' in result['statistics']
assert 'domains' in result['statistics']
@ -1393,7 +1395,7 @@ async def test_get_statistics(test_session, archive_factory):
archive4 = archive_factory() # No domain
test_session.commit()
result = await get_statistics()
result = await get_statistics(test_session)
stats = result['statistics']
# Should count all 4 archives.

7
modules/docs/api.py
View file

@ -20,8 +20,8 @@ logger = logger.getChild(__name__)
@docs_bp.get('/statistics')
@openapi.response(HTTPStatus.OK, schema.DocStatisticsResponse)
async def statistics(_: Request):
ret = get_statistics()
async def statistics(request: Request):
ret = get_statistics(request.ctx.session)
return json_response(ret, HTTPStatus.OK)
@ -44,11 +44,12 @@ doc_limit_limiter = api_param_limiter(100)
)
@openapi.response(HTTPStatus.OK, schema.DocSearchResponse)
@validate(schema.DocSearchRequest)
async def search_docs(_: Request, body: schema.DocSearchRequest):
async def search_docs(request: Request, body: schema.DocSearchRequest):
limit = doc_limit_limiter(body.limit)
offset = body.offset or 0
file_groups, total = await asyncio.to_thread(
_search_docs,
request.ctx.session,
search_str=body.search_str,
author=body.author,
subject=body.subject,

165
modules/docs/lib.py
View file

@ -10,7 +10,6 @@ from sqlalchemy.orm import Session
from wrolpi import fts
from wrolpi.collections.models import Collection
from wrolpi.collections.types import collection_type_registry
from wrolpi.db import get_db_session
from wrolpi.errors import ValidationError
from wrolpi.files.lib import split_path_stem_and_suffix, get_mimetype, cached_search_total, \
search_filter_cache_key
@ -293,24 +292,23 @@ def discover_calibre_cover(ebook_path: pathlib.Path):
return cover
def get_statistics() -> dict:
def get_statistics(session: Session) -> dict:
"""Get doc statistics."""
from .models import Doc
from wrolpi.files.models import FileGroup
with get_db_session() as session:
total = session.query(func.count(Doc.id)).scalar() or 0
total_size = session.query(func.sum(Doc.size)).scalar() or 0
total = session.query(func.count(Doc.id)).scalar() or 0
total_size = session.query(func.sum(Doc.size)).scalar() or 0
epub_count = session.query(func.count(Doc.id)).join(FileGroup).filter(
FileGroup.mimetype.startswith('application/epub')).scalar() or 0
pdf_count = session.query(func.count(Doc.id)).join(FileGroup).filter(
FileGroup.mimetype == 'application/pdf').scalar() or 0
epub_count = session.query(func.count(Doc.id)).join(FileGroup).filter(
FileGroup.mimetype.startswith('application/epub')).scalar() or 0
pdf_count = session.query(func.count(Doc.id)).join(FileGroup).filter(
FileGroup.mimetype == 'application/pdf').scalar() or 0
author_count = session.query(func.count(Collection.id)).filter(
Collection.kind == 'author').scalar() or 0
subject_count = session.query(func.count(Collection.id)).filter(
Collection.kind == 'subject').scalar() or 0
author_count = session.query(func.count(Collection.id)).filter(
Collection.kind == 'author').scalar() or 0
subject_count = session.query(func.count(Collection.id)).filter(
Collection.kind == 'subject').scalar() or 0
return dict(statistics=dict(
doc_count=total,
@ -368,7 +366,7 @@ def _doc_browse_sql(order_by: str) -> Optional[str]:
ORDER BY {order} LIMIT :limit OFFSET :offset'''
def _search_docs(search_str=None, author=None, subject=None, language=None, mimetype=None,
def _search_docs(session: Session, search_str=None, author=None, subject=None, language=None, mimetype=None,
limit=20, offset=0, order_by='published_datetime', tag_names=None, deep=False):
from .models import Doc
from wrolpi.files.models import FileGroup
@ -377,86 +375,85 @@ def _search_docs(search_str=None, author=None, subject=None, language=None, mime
# 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
with get_db_session() as session:
query = session.query(FileGroup).join(Doc, Doc.file_group_id == FileGroup.id)
query = session.query(FileGroup).join(Doc, Doc.file_group_id == FileGroup.id)
fts_sq = None
if match is not None:
fts_sq = text('SELECT rowid AS id, -rank AS ts_rank'
' FROM file_group_fts WHERE file_group_fts MATCH :fts_match') \
.bindparams(fts_match=match) \
.columns(column('id', Integer), column('ts_rank', Float)) \
.alias('fts')
query = query.join(fts_sq, fts_sq.c.id == FileGroup.id)
fts_sq = None
if match is not None:
fts_sq = text('SELECT rowid AS id, -rank AS ts_rank'
' FROM file_group_fts WHERE file_group_fts MATCH :fts_match') \
.bindparams(fts_match=match) \
.columns(column('id', Integer), column('ts_rank', Float)) \
.alias('fts')
query = query.join(fts_sq, fts_sq.c.id == FileGroup.id)
if author:
query = query.filter(FileGroup.b_text.ilike(f'%{author}%'))
if author:
query = query.filter(FileGroup.b_text.ilike(f'%{author}%'))
if subject:
query = query.filter(Doc.subject.ilike(f'%{subject}%'))
if subject:
query = query.filter(Doc.subject.ilike(f'%{subject}%'))
if language:
query = query.filter(Doc.language == language)
if language:
query = query.filter(Doc.language == language)
if mimetype:
query = query.filter(FileGroup.mimetype.startswith(mimetype))
if mimetype:
query = query.filter(FileGroup.mimetype.startswith(mimetype))
if tag_names:
from wrolpi.tags import Tag, TagFile
tagged_fg_ids = session.query(TagFile.file_group_id) \
.join(Tag, Tag.id == TagFile.tag_id) \
.filter(Tag.name.in_(tag_names)) \
.subquery()
query = query.filter(FileGroup.id.in_(tagged_fg_ids))
if tag_names:
from wrolpi.tags import Tag, TagFile
tagged_fg_ids = session.query(TagFile.file_group_id) \
.join(Tag, Tag.id == TagFile.tag_id) \
.filter(Tag.name.in_(tag_names)) \
.subquery()
query = query.filter(FileGroup.id.in_(tagged_fg_ids))
unfiltered = not (search_str or author or subject or language or mimetype or tag_names)
if unfiltered:
total = cached_search_total(
search_filter_cache_key('docs'),
lambda: session.query(func.count(Doc.id)).scalar() or 0,
)
else:
cache_key = search_filter_cache_key(
'docs', search_str=search_str, author=author, subject=subject,
language=language, mimetype=mimetype, tag_names=tag_names, deep=deep,
)
# Count before ORDER BY / LIMIT; cache by filters not page.
total = cached_search_total(cache_key, lambda q=query: q.count())
unfiltered = not (search_str or author or subject or language or mimetype or tag_names)
if unfiltered:
total = cached_search_total(
search_filter_cache_key('docs'),
lambda: session.query(func.count(Doc.id)).scalar() or 0,
)
else:
cache_key = search_filter_cache_key(
'docs', search_str=search_str, author=author, subject=subject,
language=language, mimetype=mimetype, tag_names=tag_names, deep=deep,
)
# Count before ORDER BY / LIMIT; cache by filters not page.
total = cached_search_total(cache_key, lambda q=query: q.count())
# Ordering.
if order_by == 'rank' and fts_sq is not None:
query = query.order_by(
desc(fts_sq.c.ts_rank),
desc(FileGroup.id),
)
elif order_by == 'published_datetime':
query = query.order_by(nullslast(desc(FileGroup.published_datetime)), desc(FileGroup.id))
elif order_by == 'size':
query = query.order_by(nullslast(desc(Doc.size)))
elif order_by == 'title':
query = query.order_by(nullslast(asc(FileGroup.a_text)))
else:
query = query.order_by(desc(FileGroup.id))
# Ordering.
if order_by == 'rank' and fts_sq is not None:
query = query.order_by(
desc(fts_sq.c.ts_rank),
desc(FileGroup.id),
)
elif order_by == 'published_datetime':
query = query.order_by(nullslast(desc(FileGroup.published_datetime)), desc(FileGroup.id))
elif order_by == 'size':
query = query.order_by(nullslast(desc(Doc.size)))
elif order_by == 'title':
query = query.order_by(nullslast(asc(FileGroup.a_text)))
else:
query = query.order_by(desc(FileGroup.id))
# `size` sorts on a doc column and `title` has no file_group index, so neither may fall back to `id`.
browse_sql = None
if unfiltered and order_by not in ('size', 'title'):
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 {}
file_groups = [by_id[i] for i in fg_ids]
else:
file_groups = query.offset(offset).limit(limit).all()
fg_ids = [fg.id for fg in file_groups]
results = [fg.__json__() for fg in file_groups]
# `size` sorts on a doc column and `title` has no file_group index, so neither may fall back to `id`.
browse_sql = None
if unfiltered and order_by not in ('size', 'title'):
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 {}
file_groups = [by_id[i] for i in fg_ids]
else:
file_groups = query.offset(offset).limit(limit).all()
fg_ids = [fg.id for fg in file_groups]
results = [fg.__json__() for fg in file_groups]
if search_str and fg_ids:
hints = _fetch_section_hints(session, fg_ids, search_str)
for r in results:
hint = hints.get(r['id'])
if hint:
r['section_hint'] = hint
if search_str and fg_ids:
hints = _fetch_section_hints(session, fg_ids, search_str)
for r in results:
hint = hints.get(r['id'])
if hint:
r['section_hint'] = hint
return results, total

14
modules/docs/test/test_lib.py
View file

@ -293,10 +293,10 @@ async def test_search_docs_attaches_section_hint_epub(test_session, test_directo
])
# "mullen" is only in the document body (d_text), so a deep search is required to match it.
results, total = _search_docs(search_str='mullen', mimetype='application/epub')
results, total = _search_docs(test_session, search_str='mullen', mimetype='application/epub')
assert total == 0, 'Document contents are not searched unless deep=True'
results, total = _search_docs(search_str='mullen', mimetype='application/epub', deep=True)
results, total = _search_docs(test_session, search_str='mullen', mimetype='application/epub', deep=True)
assert total == 1
assert results[0]['id'] == fg.id
hint = results[0]['section_hint']
@ -318,7 +318,7 @@ async def test_search_docs_attaches_section_hint_pdf(test_session, test_director
(3, 'Page 3', 'here is where mullen shows up'),
])
results, total = _search_docs(search_str='mullen', mimetype='application/pdf', deep=True)
results, total = _search_docs(test_session, search_str='mullen', mimetype='application/pdf', deep=True)
assert total == 1
assert results[0]['id'] == fg.id
hint = results[0]['section_hint']
@ -334,7 +334,7 @@ async def test_search_docs_no_hint_when_no_query(test_session, test_directory):
'gamma.pdf', 'application/pdf', [
(1, 'Page 1', 'whatever'),
])
results, _ = _search_docs(mimetype='application/pdf')
results, _ = _search_docs(test_session, mimetype='application/pdf')
assert results and results[0]['id'] == fg.id
assert 'section_hint' not in results[0]
@ -360,15 +360,15 @@ async def test_search_docs_browse_walks_file_group_date_index(test_session, test
assert 'TEMP B-TREE' not in plan, plan
# Newest first, NULL published dates last, same as the filtered ORM path.
results, total = _search_docs()
results, total = _search_docs(test_session, )
assert total == 3
assert [i['id'] for i in results] == [docs[2].file_group_id, docs[1].file_group_id, docs[0].file_group_id]
# Paging still works on the fast path.
results, total = _search_docs(limit=1, offset=1)
results, total = _search_docs(test_session, limit=1, offset=1)
assert total == 3
assert [i['id'] for i in results] == [docs[1].file_group_id]
# Filtered searches are unaffected.
results, total = _search_docs(mimetype='application/pdf')
results, total = _search_docs(test_session, mimetype='application/pdf')
assert total == 3

4
modules/flasher/api.py
View file

@ -19,10 +19,10 @@ flasher_bp = Blueprint('Flasher', '/api/flasher')
body=schema.FlasherSearchRequest,
)
@validate(schema.FlasherSearchRequest)
async def post_flasher_search(_: Request, body: schema.FlasherSearchRequest):
async def post_flasher_search(request: Request, body: schema.FlasherSearchRequest):
# Reading many firmware headers touches the disk; run off the event loop.
file_groups, total = await asyncio.to_thread(
lib.search_esp_firmware, body.chip, body.path, body.limit)
lib.search_esp_firmware, request.ctx.session, body.chip, body.path, body.limit)
return json_response(dict(file_groups=file_groups, totals=dict(file_groups=total)), HTTPStatus.OK)

6
modules/flasher/lib.py
View file

@ -8,6 +8,8 @@ import functools
import pathlib
from typing import List, Optional, Tuple
from sqlalchemy.orm import Session
from wrolpi.common import get_media_directory, logger
logger = logger.getChild(__name__)
@ -100,7 +102,7 @@ def _resolve_media_path(primary_path) -> pathlib.Path:
return path if path.is_absolute() else get_media_directory() / path
def search_esp_firmware(chip: Optional[str] = None, path: Optional[str] = None,
def search_esp_firmware(session: Session, chip: Optional[str] = None, path: Optional[str] = None,
limit: int = 1000) -> Tuple[List[dict], int]:
"""Search for ``.bin`` firmware, annotated with each file's detected chip/kind.
@ -114,7 +116,7 @@ def search_esp_firmware(chip: Optional[str] = None, path: Optional[str] = None,
"""
from wrolpi.files.lib import search_files
file_groups, _ = search_files(None, limit, 0, suffix='.bin', path=path)
file_groups, _ = search_files(session, None, limit, 0, suffix='.bin', path=path)
results = []
for file_group in file_groups:

8
modules/flasher/test/test_flasher.py
View file

@ -59,20 +59,20 @@ async def test_search_esp_firmware_filters_by_chip(async_client, test_session, m
return sorted(pathlib.Path(r['primary_path']).name for r in results)
# Filter to ESP32-S3: only the two S3 images, never the non-ESP file.
s3, total = search_esp_firmware(chip='ESP32-S3')
s3, total = search_esp_firmware(test_session, chip='ESP32-S3')
assert total == 2
assert names(s3) == ['s3-app.bin', 's3-factory.bin']
assert {r['esp_chip'] for r in s3} == {'ESP32-S3'}
assert {r['esp_kind'] for r in s3} == {'app', 'factory'}
# Filter to ESP32-S2: just the one.
s2, total = search_esp_firmware(chip='ESP32-S2')
s2, total = search_esp_firmware(test_session, chip='ESP32-S2')
assert total == 1
assert names(s2) == ['s2-app.bin']
# No chip: every .bin, including non-ESP parts (partition tables, littlefs, boot_app0) annotated with a null
# chip so a full flash set can still be assembled from the picker.
all_bin, total = search_esp_firmware()
all_bin, total = search_esp_firmware(test_session)
assert total == 5
assert 'littlefs.bin' in names(all_bin)
littlefs = next(r for r in all_bin if pathlib.Path(r['primary_path']).name == 'littlefs.bin')
@ -80,7 +80,7 @@ async def test_search_esp_firmware_filters_by_chip(async_client, test_session, m
assert littlefs['esp_kind'] == 'not_esp_image'
# Filtering to a specific chip still returns only that chip's ESP images (non-ESP parts are shown unfiltered).
_, total = search_esp_firmware(chip='ESP32-H2')
_, total = search_esp_firmware(test_session, chip='ESP32-H2')
assert total == 0

4
modules/videos/api.py
View file

@ -27,8 +27,8 @@ logger = logger.getChild(__name__)
@content_bp.get('/statistics')
@openapi.response(HTTPStatus.OK, schema.VideosStatisticsResponse)
async def statistics(_: Request):
ret = await lib.get_statistics()
async def statistics(request: Request):
ret = await lib.get_statistics(request.ctx.session)
return json_response(ret, HTTPStatus.OK)

4
modules/videos/channel/api.py
View file

@ -26,8 +26,8 @@ logger = logger.getChild(__name__)
@channel_bp.get('/')
@openapi.description('Get a list of all Channels')
@openapi.response(HTTPStatus.OK, schema.ChannelsResponse)
async def get_channels(_: Request):
channels = await lib.get_minimal_channels()
async def get_channels(request: Request):
channels = await lib.get_minimal_channels(request.ctx.session)
return json_response({'channels': channels})

6
modules/videos/channel/lib.py
View file

@ -9,7 +9,7 @@ from wrolpi import flags
from wrolpi.collections import Collection
from wrolpi.common import logger, \
get_media_directory, wrol_mode_check, background_task
from wrolpi.db import get_db_curs, get_db_session
from wrolpi.db import get_db_session, session_curs
from wrolpi.downloader import save_downloads_config, download_manager, Download
from wrolpi.errors import APIError, ValidationError, FileWorkerConflict
from wrolpi.events import Events
@ -23,11 +23,11 @@ from ..models import Channel, Video
logger = logger.getChild(__name__)
async def get_minimal_channels() -> List[dict]:
async def get_minimal_channels(session: Session) -> List[dict]:
"""
Get the minimum amount of information necessary about all channels.
"""
with get_db_curs() as curs:
with session_curs(session) as curs:
# Get all channels, even if they don't have videos. Also get the minimum frequency download because this is the
# one that will consume the most resources.
stmt = '''

6
modules/videos/lib.py
View file

@ -19,7 +19,7 @@ from wrolpi.captions import extract_captions
from wrolpi.common import ConfigFile, extract_domain, logger, \
escape_file_name, get_media_directory, background_task, Base, get_wrolpi_config, trim_file_name
from wrolpi.dates import Seconds, from_timestamp
from wrolpi.db import get_db_curs, get_db_session
from wrolpi.db import get_db_curs, get_db_session, session_curs
from wrolpi.downloader import Download, download_manager
from wrolpi.errors import UnknownDirectory
from wrolpi.events import Events
@ -1029,8 +1029,8 @@ def get_channel_source_id(url: str) -> str:
return channel_info.get('channel_id') or channel_info['uploader_id']
async def get_statistics():
with get_db_curs() as curs:
async def get_statistics(session: Session):
with session_curs(session) as curs:
curs.execute('''
SELECT
-- total videos

15
modules/videos/models.py
View file

@ -10,7 +10,7 @@ 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
from wrolpi.db import get_db_curs, get_db_session, session_curs
from wrolpi.downloader import Download
from wrolpi.files.lib import split_path_stem_and_suffix
from wrolpi.files.worker import file_worker
@ -177,7 +177,7 @@ class Video(ModelHelper, Base):
"""
session = Session.object_session(self)
with get_db_curs() as curs:
with session_curs(session) as curs:
if self.file_group.published_datetime:
# Get videos next to this Video's upload date.
stmt = '''
@ -964,9 +964,14 @@ class Channel(ModelHelper, Base):
d['downloads'] = self.downloads
return d
def get_statistics(self):
"""Get statistics about this channel."""
with get_db_curs() as curs:
def get_statistics(self) -> dict:
"""Get statistics about this channel.
Read on the Channel's own Session, so during a request this uses the request's connection."""
session = Session.object_session(self)
if session is None:
raise RuntimeError(f'{self} is not in a session!')
with session_curs(session) as curs:
stmt = '''
SELECT SUM(size) AS "size",
MAX(size) AS "largest_video",

2
modules/videos/test/test_api.py
View file

@ -680,7 +680,7 @@ async def test_search_videos_default_page_pagination(test_session, video_factory
# The fast path must return exactly what the generic query returns for the same request:
# same rows, same order (NULL placement, id tiebreaker, interleaved mimetypes), same paging.
def ids(order, limit, offset):
results, total = lib.search_videos(order=order, limit=limit, offset=offset)
results, total = lib.search_videos(test_session, order=order, limit=limit, offset=offset)
return [i['id'] for i in results], total
for order in lib.INDEXED_DATE_ORDERS:

4
modules/videos/test/test_lib.py
View file

@ -167,7 +167,7 @@ def test_validate_video_directory_beats_info_json_channel(test_session, test_dir
@pytest.mark.asyncio
async def test_get_statistics(test_session, video_factory, channel_factory):
# Can get statistics in empty DB.
await get_statistics()
await get_statistics(test_session)
channel1 = channel_factory()
channel2 = channel_factory()
@ -176,7 +176,7 @@ async def test_get_statistics(test_session, video_factory, channel_factory):
video_factory(channel_id=channel2.id)
video_factory()
result = await get_statistics()
result = await get_statistics(test_session)
assert 'statistics' in result
assert 'videos' in result['statistics']
assert 'channels' in result['statistics']

24
modules/videos/test/test_video_edit.py
View file

@ -20,7 +20,8 @@ def test_update_video_title(test_session, test_directory, video_factory):
test_session.commit()
assert video.file_group.title == 'Old Title'
update_video(video.file_group_id, ' New Title ')
update_video(test_session, video.file_group_id, ' New Title ')
test_session.commit()
test_session.expire_all()
video = Video.find_by_file_group_id(test_session, video.file_group_id)
@ -39,7 +40,8 @@ def test_update_video_title_creates_info_json(test_session, test_directory, vide
test_session.commit()
assert video.info_json_path is None
update_video(video.file_group_id, 'Brand New')
update_video(test_session, video.file_group_id, 'Brand New')
test_session.commit()
test_session.expire_all()
video = Video.find_by_file_group_id(test_session, video.file_group_id)
@ -57,7 +59,8 @@ def test_update_video_description(test_session, test_directory, video_factory):
test_session.commit()
assert video.file_group.c_text == 'old words'
update_video(video.file_group_id, description='new words\nsecond line')
update_video(test_session, video.file_group_id, description='new words\nsecond line')
test_session.commit()
test_session.expire_all()
video = Video.find_by_file_group_id(test_session, video.file_group_id)
@ -68,7 +71,8 @@ def test_update_video_description(test_session, test_directory, video_factory):
assert info_json['description'] == 'old words', 'The original is kept'
assert info_json['wrolpi']['custom_description'] == 'new words\nsecond line'
update_video(video.file_group_id, description='')
update_video(test_session, video.file_group_id, description='')
test_session.commit()
test_session.expire_all()
video = Video.find_by_file_group_id(test_session, video.file_group_id)
assert video.get_description() is None
@ -80,14 +84,14 @@ def test_update_video_nothing(test_session, video_factory):
video = video_factory(with_info_json={'title': 'Old'})
test_session.commit()
with pytest.raises(ValidationError):
update_video(video.file_group_id)
update_video(test_session, video.file_group_id)
def test_update_video_title_empty(test_session, video_factory):
video = video_factory(with_info_json={'title': 'Old'})
test_session.commit()
with pytest.raises(ValidationError):
update_video(video.file_group_id, ' ')
update_video(test_session, video.file_group_id, ' ')
test_session.expire_all()
assert Video.find_by_file_group_id(test_session, video.file_group_id).file_group.title == 'Old'
@ -108,7 +112,8 @@ async def test_custom_title_survives_refresh(test_session, test_directory, async
video = video_factory(with_info_json={'title': 'Original'})
test_session.commit()
update_video(video.file_group_id, title='Custom', description='custom words')
update_video(test_session, video.file_group_id, title='Custom', description='custom words')
test_session.commit()
video_path = video.video_path
file_group_id = video.file_group_id
@ -174,7 +179,7 @@ async def test_update_video_wrol_mode(test_session, video_factory, wrol_mode_fix
await wrol_mode_fixture(True)
try:
with pytest.raises(WROLModeEnabled):
update_video(video.file_group_id, title='Nope')
update_video(test_session, video.file_group_id, title='Nope')
finally:
await wrol_mode_fixture(False)
@ -194,7 +199,8 @@ async def test_edit_creates_info_json_that_survives_refresh(test_session, test_d
video_path = video.video_path
file_group_id = video.file_group_id
update_video(file_group_id, title='Custom', description='my words')
update_video(test_session, file_group_id, title='Custom', description='my words')
test_session.commit()
test_session.expire_all()
video = Video.find_by_file_group_id(test_session, file_group_id)
assert video.info_json_path and video.info_json_path.is_file()

30
modules/videos/video/api.py
View file

@ -25,7 +25,9 @@ logger = logger.getChild(__name__)
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
def video_get(request: Request, file_group_id: int):
skip_viewed = request.args.get('skip_viewed', '').lower() == 'true'
video, previous_video, next_video = lib.get_video_for_app(file_group_id, skip_viewed=skip_viewed)
# Marking the Video viewed is a write.
with request.ctx.db.write() as session:
video, previous_video, next_video = lib.get_video_for_app(session, file_group_id, skip_viewed=skip_viewed)
return json_response({'file_group': video, 'prev': previous_video, 'next': next_video})
@ -37,9 +39,10 @@ def video_get(request: Request, file_group_id: int):
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
@validate(schema.VideoUpdateRequest)
@wrol_mode_check
async def video_update(_: Request, file_group_id: int, body: schema.VideoUpdateRequest):
lib.update_video(file_group_id, title=body.title, description=body.description)
video, previous_video, next_video = lib.get_video_for_app(file_group_id)
async def video_update(request: Request, file_group_id: int, body: schema.VideoUpdateRequest):
with request.ctx.db.write() as session:
lib.update_video(session, file_group_id, title=body.title, description=body.description)
video, previous_video, next_video = lib.get_video_for_app(session, file_group_id)
return json_response({'file_group': video, 'prev': previous_video, 'next': next_video})
@ -47,8 +50,8 @@ async def video_update(_: Request, file_group_id: int, body: schema.VideoUpdateR
@openapi.description('Get Video comments')
@openapi.response(HTTPStatus.OK, schema.VideoCommentsResponse)
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
def video_get_comments(_: Request, file_group_id: int):
video = lib.get_video(file_group_id)
def video_get_comments(request: Request, file_group_id: int):
video = lib.get_video(request.ctx.session, file_group_id)
return json_response({'comments': video.get_comments()})
@ -56,8 +59,8 @@ def video_get_comments(_: Request, file_group_id: int):
@openapi.description('Get Video description')
@openapi.response(HTTPStatus.OK, schema.VideoDescriptionResponse)
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
def video_get_description(_: Request, file_group_id: int):
video = lib.get_video(file_group_id)
def video_get_description(request: Request, file_group_id: int):
video = lib.get_video(request.ctx.session, file_group_id)
return json_response({'description': video.get_description()})
@ -65,8 +68,8 @@ def video_get_description(_: Request, file_group_id: int):
@openapi.description('Get Video captions')
@openapi.response(HTTPStatus.OK, schema.VideoCaptionsResponse)
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
def video_get_captions(_: Request, file_group_id: int):
video = lib.get_video(file_group_id)
def video_get_captions(request: Request, file_group_id: int):
video = lib.get_video(request.ctx.session, file_group_id)
return json_response({'captions': video.get_caption_chunks()})
@ -78,13 +81,13 @@ def video_get_captions(_: Request, file_group_id: int):
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
@validate(schema.VideoTranscodeRequest)
@wrol_mode_check
async def video_transcode(_: Request, file_group_id: int, body: schema.VideoTranscodeRequest):
async def video_transcode(request: Request, file_group_id: int, body: schema.VideoTranscodeRequest):
try:
validate_transcode_request(body.video_codec, body.audio_codec, body.container, body.fragmented)
except ValueError as e:
raise InvalidJob(str(e))
# Raises UnknownVideo (404) before anything is queued.
video = lib.get_video(file_group_id)
video = lib.get_video(request.ctx.session, file_group_id)
is_audio = (video.file_group.mimetype or '').startswith('audio/')
if is_audio and body.video_codec and body.video_codec != REMOVE_VIDEO:
raise InvalidJob(f'This is an audio file; it has no video stream to transcode to {body.video_codec}')
@ -117,13 +120,14 @@ async def video_transcode(_: Request, file_group_id: int, body: schema.VideoTran
@openapi.response(HTTPStatus.OK, schema.VideoSearchResponse)
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
@validate(schema.VideoSearchRequest)
async def search_videos(_: Request, body: schema.VideoSearchRequest):
async def search_videos(request: Request, body: schema.VideoSearchRequest):
if body.order_by not in lib.VIDEO_ORDERS:
raise InvalidOrderBy('Invalid order by')
# Synchronous SQLite query; keep it off the event loop.
file_groups, videos_total = await asyncio.to_thread(
lib.search_videos,
request.ctx.session,
body.search_str,
body.offset,
body.limit,

80
modules/videos/video/lib.py
View file

@ -11,7 +11,7 @@ from modules.videos.models import Video, Channel
from wrolpi import fts
from wrolpi.common import logger, limit_concurrent, wrol_mode_check
from wrolpi.dates import now
from wrolpi.db import get_db_session, get_db_curs
from wrolpi.db import get_db_session, session_curs
from wrolpi.downloader import download_manager
from wrolpi.files.lib import handle_file_group_search_results, cached_search_total, \
search_filter_cache_key, count_file_groups
@ -24,34 +24,32 @@ from ..lib import get_yt_dlp_http_headers, get_yt_dlp_sleep_opts
logger.getChild(__name__)
def get_video_for_app(file_group_id: int, skip_viewed: bool = False) -> Tuple[dict, Optional[dict], Optional[dict]]:
def get_video_for_app(session: Session, file_group_id: int, skip_viewed: bool = False) \
-> Tuple[dict, Optional[dict], Optional[dict]]:
"""
Get a Video by its FileGroup ID, with its prev/next videos. Mark the Video as viewed.
Get a Video by its FileGroup ID, with its prev/next videos. Mark the Video as viewed (a write; the
caller commits).
"""
with get_db_session(commit=True) as session:
video = Video.find_by_file_group_id(session, file_group_id)
if not skip_viewed:
video.file_group.set_viewed()
previous_video, next_video = video.get_surrounding_videos()
video = video.__json__()
previous_video = previous_video.__json__() if previous_video and previous_video.file_group else None
next_video = next_video.__json__() if next_video and next_video.file_group else None
video = Video.find_by_file_group_id(session, file_group_id)
if not skip_viewed:
video.file_group.set_viewed()
previous_video, next_video = video.get_surrounding_videos()
video = video.__json__()
previous_video = previous_video.__json__() if previous_video and previous_video.file_group else None
next_video = next_video.__json__() if next_video and next_video.file_group else None
return video, previous_video, next_video
def get_video(file_group_id: int) -> Video:
def get_video(session: Session, file_group_id: int) -> Video:
"""
Get a Video by its FileGroup ID.
"""
with get_db_session() as session:
video = Video.find_by_file_group_id(session, file_group_id)
return video
return Video.find_by_file_group_id(session, file_group_id)
@wrol_mode_check
def update_video(file_group_id: int, title: str = None, description: str = None) -> Video:
def update_video(session: Session, file_group_id: int, title: str = None, description: str = None) -> Video:
"""Edit a Video's details. The files are the source of truth, so the change is written to the
Video's .info.json (created if missing) and the Video is validated again from it. Only the
fields given (not None) are changed; an empty description clears it.
@ -60,6 +58,8 @@ def update_video(file_group_id: int, title: str = None, description: str = None)
over yt-dlp's own keys: the original title and description stay in the file, and a metadata
re-download carries the section over.
The caller commits.
@raise UnknownVideo: if the Video can not be found
@raise ValidationError: when nothing is given, or the title is empty
"""
@ -77,17 +77,16 @@ def update_video(file_group_id: int, title: str = None, description: str = None)
if description is not None:
changes['custom_description'] = description
with get_db_session(commit=True) as session:
video = Video.find_by_file_group_id(session, file_group_id)
if video.info_json_path is None or not video.info_json_path.is_file():
# No info json yet: create one holding only the wrolpi section.
video.replace_info_json({'wrolpi': changes}, clean=False)
else:
video.file_group.update_wrolpi_json(changes)
# Re-derive the title (and search text) from the file just written.
video.validate(session)
video.flush()
return video
video = Video.find_by_file_group_id(session, file_group_id)
if video.info_json_path is None or not video.info_json_path.is_file():
# No info json yet: create one holding only the wrolpi section.
video.replace_info_json({'wrolpi': changes}, clean=False)
else:
video.file_group.update_wrolpi_json(changes)
# Re-derive the title (and search text) from the file just written.
video.validate(session)
video.flush()
return video
VIDEO_ORDERS = {
@ -132,10 +131,10 @@ INDEXED_DATE_ORDERS = ('published_datetime', '-published_datetime')
VIDEO_MIMETYPE_PREFIXES = ('video/', 'audio/')
def _video_mimetypes() -> List[str]:
def _video_mimetypes(session: Session) -> List[str]:
"""Every distinct video/* and audio/* mimetype in the library (a handful; served by the mimetype index)."""
mimetypes = []
with get_db_curs() as curs:
with session_curs(session) as curs:
for prefix in VIDEO_MIMETYPE_PREFIXES:
# '0' is the character after '/', so this is the half-open range of the prefix.
curs.execute('SELECT DISTINCT mimetype FROM file_group WHERE mimetype >= ? AND mimetype < ?',
@ -144,17 +143,17 @@ def _video_mimetypes() -> List[str]:
return mimetypes
def _count_videos() -> int:
def _count_videos(session: Session) -> int:
ranges = ' OR '.join(f'(mimetype >= :p{i} AND mimetype < :e{i})' for i in range(len(VIDEO_MIMETYPE_PREFIXES)))
params = dict()
for i, prefix in enumerate(VIDEO_MIMETYPE_PREFIXES):
params[f'p{i}'], params[f'e{i}'] = prefix, prefix[:-1] + '0'
with get_db_curs() as curs:
with session_curs(session) as curs:
curs.execute(f'SELECT COUNT(*) FROM file_group WHERE {ranges}', params)
return curs.fetchone()[0]
def _search_videos_by_date(order: str, limit: int, offset: int) -> Tuple[List[dict], int]:
def _search_videos_by_date(session: Session, order: str, limit: int, offset: int) -> Tuple[List[dict], int]:
"""The unfiltered Videos page, ordered by date, without visiting every video.
The generic query (`mimetype LIKE 'video/%' OR ...`) plus a windowed COUNT used to visit every
@ -163,7 +162,7 @@ def _search_videos_by_date(order: str, limit: int, offset: int) -> Tuple[List[di
then merged. Results must be identical to the generic query's (same ORDER BY, including the
id tiebreaker and NULL placement). The total is a separate, cached count."""
order_by = VIDEO_ORDERS[order]
mimetypes = _video_mimetypes()
mimetypes = _video_mimetypes(session)
if not mimetypes:
return [], 0
@ -183,11 +182,12 @@ def _search_videos_by_date(order: str, limit: int, offset: int) -> Tuple[List[di
'''.strip()
logger.debug(f'{stmt} {params}')
total = cached_search_total(search_filter_cache_key('videos'), _count_videos)
return handle_file_group_search_results(stmt, params, total=total)
total = cached_search_total(search_filter_cache_key('videos'), lambda: _count_videos(session))
return handle_file_group_search_results(session, stmt, params, total=total)
def search_videos(
session: Session,
search_str: str = None,
offset: int = None,
limit: int = VIDEO_QUERY_LIMIT,
@ -212,7 +212,7 @@ def search_videos(
offset = int(offset) if offset else 0
if not search_str and not tag_names and not channel_id and not censored and order in INDEXED_DATE_ORDERS:
# The Videos page with no filters, in date order.
return _search_videos_by_date(order, limit, offset)
return _search_videos_by_date(session, order, limit, offset)
params = dict(search_str=search_str, offset=offset)
if channel_id:
@ -280,7 +280,7 @@ def search_videos(
unfiltered = (not search_str and not tag_names and not channel_id and not censored
and not null_filter)
if unfiltered:
total = cached_search_total(search_filter_cache_key('videos'), _count_videos)
total = cached_search_total(search_filter_cache_key('videos'), lambda: _count_videos(session))
else:
if fts_search:
fts_count = fts.file_group_search(search_str, deep=deep)
@ -303,9 +303,9 @@ def search_videos(
'videos', search_str=search_str, channel_id=channel_id, tag_names=tag_names,
censored=censored, deep=deep, null_filter=null_filter,
)
total = cached_search_total(cache_key, lambda: count_file_groups(count_stmt, count_params))
total = cached_search_total(cache_key, lambda: count_file_groups(session, count_stmt, count_params))
results, total = handle_file_group_search_results(stmt, params, total=total)
results, total = handle_file_group_search_results(session, stmt, params, total=total)
return results, total

12
modules/zim/lib.py
View file

@ -22,7 +22,7 @@ from wrolpi.common import register_modeler, logger, extract_html_text, extract_h
from wrolpi.switches import register_switch_handler
from wrolpi.dates import Seconds
from wrolpi.events import Events
from wrolpi.db import get_db_session, get_db_curs
from wrolpi.db import get_db_session, session_curs
from wrolpi.downloader import DownloadFrequency
from wrolpi.files.lib import split_file_name_words
from wrolpi.files.modeler import run_modeler_loop, skip_clause
@ -112,7 +112,7 @@ async def zim_modeler(progress_callback: Callable[[int], None] = None):
)
def get_all_entries_tags():
def get_all_entries_tags(session: Session):
"""Returns a dict of dicts which contain the names of all Tags of each Entry's path.
Example:
@ -130,7 +130,7 @@ def get_all_entries_tags():
LEFT JOIN tag_zim tz on t.id = tz.tag_id
GROUP BY 1, 2 \
'''
with get_db_curs() as curs:
with session_curs(session) as curs:
curs.execute(stmt)
entries = dict()
for zim_id, zim_entry, tag_names in curs.fetchall():
@ -144,8 +144,8 @@ def get_all_entries_tags():
return entries
def get_entries_tags(paths: List[str], zim_id: int):
all_entries_tags = get_all_entries_tags()
def get_entries_tags(session: Session, paths: List[str], zim_id: int):
all_entries_tags = get_all_entries_tags(session)
entries = dict()
for path in paths:
zim_tag_entries = all_entries_tags.get(zim_id, dict())
@ -187,7 +187,7 @@ def headline_zim(session: Session, search_str: str, zim_id: int, tag_names: List
articles = [bytes(i.get_item().content).decode('UTF-8') for i in entries]
articles = [extract_html_text(i) for i in articles]
entries_tag_names = get_entries_tags([i.path for i in entries], zim_id)
entries_tag_names = get_entries_tags(session, [i.path for i in entries], zim_id)
# Always get headlines even if there is no search_str. Non-matches get the leading text of each article.
headlines = extract_headlines(articles, search_str)

4
modules/zim/models.py
View file

@ -14,7 +14,7 @@ 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 get_db_curs
from wrolpi.db import session_curs
from wrolpi.downloader import Download, download_manager
from wrolpi.files.models import FileGroup
from wrolpi.media_path import MediaPathType
@ -249,7 +249,7 @@ class Zim(Base, ModelHelper):
@staticmethod
def _entries_with_tags_process(limit: int, offset: int, params: dict, session: Session, tags_sub_select: str):
with get_db_curs() as curs:
with session_curs(session) as curs:
curs.execute(tags_sub_select, params)
# sqlite3.Row is not a tuple; convert for the `tuple_(...).in_()` below.
zim_ids_entries: List[Tuple[int, str]] = [tuple(row) for row in curs.fetchall()]

2
modules/zim/test/test_lib.py
View file

@ -64,7 +64,7 @@ async def test_zim_get_entries_tags(async_client, test_session, test_zim, tag_fa
test_zim.tag_entry(test_session, tag1.name, 'two')
test_session.commit()
assert lib.get_entries_tags(['one', 'two', 'home'], 1) \
assert lib.get_entries_tags(test_session, ['one', 'two', 'home'], 1) \
== dict(one=['tag1', 'tag2'], two=['tag1'], home=[])

16
wrolpi/common.py
View file

@ -2825,10 +2825,9 @@ def extract_headlines(entries: List[str], search_str: str) -> List[Tuple[str, fl
return fts.headline_texts(entries, search_str, tokens=8)
async def search_other_estimates(tag_names: List[str]) -> dict:
async def search_other_estimates(session: Session, tag_names: List[str]) -> dict:
"""Estimate other things that are Tagged."""
from sqlalchemy import func
from wrolpi.db import get_db_session
from wrolpi.collections.models import Collection
from wrolpi.tags import Tag
from modules.videos.models import Channel
@ -2838,13 +2837,12 @@ async def search_other_estimates(tag_names: List[str]) -> dict:
channel_count=0,
)
with get_db_session() as session:
# TODO handle multiple tags
channel_count = session.query(func.count(Channel.id)) \
.join(Collection, Collection.id == Channel.collection_id) \
.join(Tag, Tag.id == Collection.tag_id) \
.filter(Tag.name == tag_names[0]) \
.scalar()
# TODO handle multiple tags
channel_count = session.query(func.count(Channel.id)) \
.join(Collection, Collection.id == Channel.collection_id) \
.join(Tag, Tag.id == Collection.tag_id) \
.filter(Tag.name == tag_names[0]) \
.scalar()
others = dict(
channel_count=channel_count,

36
wrolpi/db.py
View file

@ -13,7 +13,7 @@ import sqlite3
import threading
import types
from contextlib import contextmanager
from typing import Tuple, List, Union, Type, Generator, Any
from typing import Tuple, List, Union, Type, Generator, Any, ContextManager
import sqlalchemy
import sqlalchemy.exc
@ -317,6 +317,23 @@ def get_db_curs(commit: bool = False) -> Generator[sqlite3.Cursor, Any, None]:
connection.close()
@contextmanager
def session_curs(session: Session) -> Generator[sqlite3.Cursor, Any, None]:
"""A raw `sqlite3.Cursor` on `session`'s connection, inside the Session's transaction.
Use it for raw SQL in code that is handed a Session, so the SQL shares that Session's connection
instead of opening another (as `get_db_curs` does). Pending ORM changes are flushed first, so raw
SQL sees them. Writes through the cursor commit or roll back with the Session. Rows are
`sqlite3.Row`."""
session.flush()
curs = session.connection().connection.cursor()
curs.row_factory = sqlite3.Row
try:
yield curs
finally:
curs.close()
def _session_has_connection(session: Session) -> bool:
"""True if the Session's transaction has already connected (and so already emitted BEGIN)."""
# SQLAlchemy 1.3 has no public API for this; 2.0 replaces it with `session.in_transaction()`.
@ -345,19 +362,10 @@ class RequestDB:
self.engine, self.session = _get_request_session()
_refuse_non_test_engine(self.engine)
@contextmanager
def curs(self) -> Generator[sqlite3.Cursor, Any, None]:
"""A raw `sqlite3.Cursor` on this request's connection, inside the Session's transaction.
Pending ORM changes are flushed first, so raw SQL sees them. Writes through the cursor commit
or roll back with the request (or with the enclosing `write()` block). Rows are `sqlite3.Row`."""
self.session.flush()
curs = self.session.connection().connection.cursor()
curs.row_factory = sqlite3.Row
try:
yield curs
finally:
curs.close()
def curs(self) -> ContextManager[sqlite3.Cursor]:
"""A raw `sqlite3.Cursor` on this request's connection (see `session_curs`). Its writes commit or
roll back with the request, or with the enclosing `write()` block."""
return session_curs(self.session)
@contextmanager
def write(self) -> Generator[Session, Any, None]:

41
wrolpi/downloader.py
View file

@ -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, get_db_curs
from wrolpi.db import get_db_session, session_curs
from wrolpi.errors import InvalidDownload, UnrecoverableDownloadError, BotBlockedDownloadError, UnknownDownload, \
ValidationError, DownloadError
from wrolpi.events import Events
@ -1307,16 +1307,15 @@ class DownloadManager:
raise TimeoutError('Downloads never finished!')
@staticmethod
def retry_downloads(reset_attempts: bool = False):
"""Set any incomplete Downloads to `new` so they will be retried.
def retry_downloads(session: Session, reset_attempts: bool = False):
"""Set any incomplete Downloads to `new` so they will be retried. The caller commits.
@param reset_attempts: Will set `download.attempts` to 0 if True.
"""
values = dict(status=DownloadStatus.new, attempts=0) if reset_attempts else dict(status=DownloadStatus.new)
with get_db_session(commit=True) as session:
session.query(Download) \
.filter(Download.status.in_((DownloadStatus.pending, DownloadStatus.deferred, DownloadStatus.failed))) \
.update(values, synchronize_session=False)
session.query(Download) \
.filter(Download.status.in_((DownloadStatus.pending, DownloadStatus.deferred, DownloadStatus.failed))) \
.update(values, synchronize_session=False)
def get_new_downloads(self, session: Session) -> Generator[Download, None, None]:
"""
@ -1450,17 +1449,16 @@ class DownloadManager:
return download
def kill_download(self, download_id: int):
"""Fail a Download. If it is pending, kill the Downloader so the download stops."""
def kill_download(self, session: Session, download_id: int):
"""Fail a Download. If it is pending, kill the Downloader so the download stops. The caller commits."""
logger.info(f'Killing Download: {download_id}')
download_manager_data = api_app.shared_ctx.download_manager_data.copy()
download_manager_data['killed_downloads'] = download_manager_data['killed_downloads'] + [download_id, ]
api_app.shared_ctx.download_manager_data.update(download_manager_data)
with get_db_session(commit=True) as session:
if download := Download.get_by_id(session, download_id):
download.error = 'User stopped this download'
download.fail()
if download := Download.get_by_id(session, download_id):
download.error = 'User stopped this download'
download.fail()
@staticmethod
def unkill_download(download_id: int):
@ -1518,10 +1516,10 @@ class DownloadManager:
download[key] = parse_db_datetime(download[key])
return download
def get_fe_downloads(self):
def get_fe_downloads(self, session: Session):
"""Get downloads for the Frontend. Uses raw SQL for faster result."""
# Use custom SQL because SQLAlchemy is slow.
with get_db_curs() as curs:
with session_curs(session) as curs:
stmt = f'''
SELECT
collection_id,
@ -1598,18 +1596,17 @@ class DownloadManager:
)
return data
def get_summary(self) -> dict:
def get_summary(self, session: Session) -> dict:
"""
Get a summary of what Downloads are happening as well as the status of the DownloadManager.
"""
from sqlalchemy import func
with get_db_session() as session:
pending_downloads, recurring_downloads = session.query(
func.count(Download.id).filter(Download.status == DownloadStatus.pending),
func.count(Download.id).filter(Download.frequency != None), # noqa
).one()
daily_limit_reached = self.daily_limit_reached(session)
pending_downloads, recurring_downloads = session.query(
func.count(Download.id).filter(Download.status == DownloadStatus.pending),
func.count(Download.id).filter(Download.frequency != None), # noqa
).one()
daily_limit_reached = self.daily_limit_reached(session)
summary = dict(
pending=pending_downloads,
recurring=recurring_downloads,

9
wrolpi/files/api.py
View file

@ -151,10 +151,11 @@ async def worker_status(request: Request):
body=schema.FilesSearchRequest,
)
@validate(schema.FilesSearchRequest)
async def post_search_files(_: Request, body: schema.FilesSearchRequest):
async def post_search_files(request: Request, body: schema.FilesSearchRequest):
with timer('Searching all files', 'info', logger__=logger):
# The handler awaits the thread, so the request's Session has one user at a time.
file_groups, total = await asyncio.to_thread(
lib.search_files, body.search_str, body.limit, body.offset, body.mimetypes, body.model,
lib.search_files, request.ctx.session, body.search_str, body.limit, body.offset, body.mimetypes, body.model,
body.tag_names, body.headline, body.months, body.from_year, body.to_year,
body.any_tag, body.order, body.url, body.suffix, body.path, body.deep)
return json_response(dict(file_groups=file_groups, totals=dict(file_groups=total)))
@ -524,8 +525,8 @@ async def post_unignore_directory(request: Request, body: schema.Directory):
body=schema.BulkTagPreviewRequest,
)
@validate(schema.BulkTagPreviewRequest)
async def post_bulk_tag_preview(_: Request, body: schema.BulkTagPreviewRequest):
preview = lib.get_bulk_tag_preview(body.paths)
async def post_bulk_tag_preview(request: Request, body: schema.BulkTagPreviewRequest):
preview = lib.get_bulk_tag_preview(request.ctx.session, body.paths)
return json_response(preview.__json__())

169
wrolpi/files/lib.py
View file

@ -30,7 +30,7 @@ from wrolpi.common import get_media_directory, wrol_mode_check, logger, \
get_wrolpi_config, \
unique_by_predicate, get_paths_in_media_directory, TRACE_LEVEL, get_relative_to_media_directory, strip_surrogates
from wrolpi.dates import now, from_timestamp, months_selector_to_where, date_range_to_where
from wrolpi.db import get_db_session, get_db_curs, values_clause
from wrolpi.db import get_db_session, get_db_curs, values_clause, session_curs
from wrolpi.downloader import download_manager, Download
from wrolpi.errors import InvalidFile, UnknownDirectory, UnknownFile, UnknownTag, FileConflict, FileGroupIsTagged, \
NoPrimaryFile, InvalidDirectory, IgnoredDirectoryError, UnsupportedArchive, InvalidArchiveMember
@ -1149,15 +1149,15 @@ def search_filter_cache_key(corpus: str, **filters) -> str:
return json.dumps(payload, sort_keys=True, default=str)
def count_file_groups(statement: str, params: dict) -> int:
def count_file_groups(session: Session, statement: str, params: dict) -> int:
"""Run a `SELECT COUNT(*) AS total ...` statement and return the integer."""
with get_db_curs() as curs:
with session_curs(session) as curs:
curs.execute(statement, params)
row = curs.fetchone()
return int(row['total'])
def search_files(search_str: str, limit: int, offset: int, mimetypes: List[str] = None, model: str = None,
def search_files(session: Session, search_str: str, limit: int, offset: int, mimetypes: List[str] = None, model: str = None,
tag_names: List[str] = None, headline: bool = False, months: List[int] = None,
from_year: int = None, to_year: int = None, any_tag: bool = False, order: str = None,
url: str = None, suffix: str = None, path: str = None, deep: bool = False) -> \
@ -1294,12 +1294,13 @@ def search_files(search_str: str, limit: int, offset: int, mimetypes: List[str]
any_tag=any_tag, url=url, suffix=suffix, path=path, deep=deep,
viewed_only=viewed_only,
)
total = cached_search_total(cache_key, lambda: count_file_groups(count_stmt, count_params))
results, total = handle_file_group_search_results(stmt, params, total=total)
total = cached_search_total(cache_key, lambda: count_file_groups(session, count_stmt, count_params))
results, total = handle_file_group_search_results(session, stmt, params, total=total)
return results, total
def handle_file_group_search_results(statement: str, params: dict, total: int = None) -> Tuple[List[dict], int]:
def handle_file_group_search_results(session: Session, statement: str, params: dict, total: int = None) \
-> Tuple[List[dict], int]:
"""
Execute the provided SQL statement and fetch the Files returned.
@ -1310,7 +1311,7 @@ def handle_file_group_search_results(statement: str, params: dict, total: int =
See: `search`
"""
with get_db_curs() as curs:
with session_curs(session) as curs:
curs.execute(statement, params)
results = [dict(i) for i in curs.fetchall()]
if total is None:
@ -1327,56 +1328,55 @@ def handle_file_group_search_results(statement: str, params: dict, total: int =
d_headline=i.get('d_headline'),
) for i in results]
with get_db_session() as session:
from modules.archive.models import Archive
from modules.videos.models import Video
results = session.query(FileGroup, Video, Archive) \
.filter(FileGroup.id.in_(ordered_ids)) \
.outerjoin(Video, Video.file_group_id == FileGroup.id) \
.outerjoin(Archive, Archive.file_group_id == FileGroup.id)
# Order FileGroups by their location in ordered_ids. Use dict for O(1) lookups instead of O(n) list.index().
id_to_order = {id_: idx for idx, id_ in enumerate(ordered_ids)}
file_groups = sorted(results, key=lambda i: id_to_order.get(i[0].id, 0))
results = list()
for extra, (file_group, video, archive) in zip_longest(extras, file_groups):
if video:
results.append(video.__json__())
elif archive:
results.append(archive.__json__())
else:
results.append(file_group.__json__())
# Preserve the ts_ranks, if any.
if extra:
results[-1]['ts_rank'] = extra['ts_rank']
results[-1]['title_headline'] = extra['title_headline']
results[-1]['b_headline'] = extra['b_headline']
results[-1]['c_headline'] = extra['c_headline']
results[-1]['d_headline'] = extra['d_headline']
from modules.archive.models import Archive
from modules.videos.models import Video
results = session.query(FileGroup, Video, Archive) \
.filter(FileGroup.id.in_(ordered_ids)) \
.outerjoin(Video, Video.file_group_id == FileGroup.id) \
.outerjoin(Archive, Archive.file_group_id == FileGroup.id)
# Order FileGroups by their location in ordered_ids. Use dict for O(1) lookups instead of O(n) list.index().
id_to_order = {id_: idx for idx, id_ in enumerate(ordered_ids)}
file_groups = sorted(results, key=lambda i: id_to_order.get(i[0].id, 0))
results = list()
for extra, (file_group, video, archive) in zip_longest(extras, file_groups):
if video:
results.append(video.__json__())
elif archive:
results.append(archive.__json__())
else:
results.append(file_group.__json__())
# Preserve the ts_ranks, if any.
if extra:
results[-1]['ts_rank'] = extra['ts_rank']
results[-1]['title_headline'] = extra['title_headline']
results[-1]['b_headline'] = extra['b_headline']
results[-1]['c_headline'] = extra['c_headline']
results[-1]['d_headline'] = extra['d_headline']
search_str = params.get('search_str') if isinstance(params, dict) else None
search_str = params.get('search_str') if isinstance(params, dict) else None
# When headlines were requested, compute the title headlines in Python (the FTS5 snippet
# function only works on indexed columns; `title` is not one).
if search_str and any(extra and extra.get('headlines') for extra in extras):
from wrolpi import fts
titles = [result.get('title') or '' for result in results]
for result, (headline_text, rank) in zip(results, fts.headline_texts(titles, search_str)):
if rank > 0:
result['title_headline'] = headline_text
# When headlines were requested, compute the title headlines in Python (the FTS5 snippet
# function only works on indexed columns; `title` is not one).
if search_str and any(extra and extra.get('headlines') for extra in extras):
from wrolpi import fts
titles = [result.get('title') or '' for result in results]
for result, (headline_text, rank) in zip(results, fts.headline_texts(titles, search_str)):
if rank > 0:
result['title_headline'] = headline_text
# When searching by text, attach a `section_hint` to any doc-modeled result so
# the UI can deep-link into the matching EPUB chapter or PDF page.
if search_str:
from modules.docs.lib import _fetch_section_hints
# `r['id']` is the FileGroup primary key; _fetch_section_hints filters
# on `file_group_id`, so this list is file-group ids (not doc.id).
fg_ids = [r['id'] for r in results if r.get('model') == 'doc']
if fg_ids:
hints = _fetch_section_hints(session, fg_ids, search_str)
for r in results:
hint = hints.get(r['id'])
if hint:
r['section_hint'] = hint
# When searching by text, attach a `section_hint` to any doc-modeled result so
# the UI can deep-link into the matching EPUB chapter or PDF page.
if search_str:
from modules.docs.lib import _fetch_section_hints
# `r['id']` is the FileGroup primary key; _fetch_section_hints filters
# on `file_group_id`, so this list is file-group ids (not doc.id).
fg_ids = [r['id'] for r in results if r.get('model') == 'doc']
if fg_ids:
hints = _fetch_section_hints(session, fg_ids, search_str)
for r in results:
hint = hints.get(r['id'])
if hint:
r['section_hint'] = hint
return results, total
@ -1458,8 +1458,8 @@ def split_file_name_words(name: str) -> str:
return name
def get_file_statistics():
with get_db_curs() as curs:
def get_file_statistics(session: Session):
with session_curs(session) as curs:
curs.execute('''
SELECT
-- All items in file_group.files are real individual files.
@ -1573,7 +1573,7 @@ def mimetypes_to_sql_wheres(wheres: List[str], params: dict, mimetypes: List[str
return wheres, params
async def search_file_suggestion_count(search_str: str, tag_names: List[str], mimetypes: List[str],
async def search_file_suggestion_count(session: Session, search_str: str, tag_names: List[str], mimetypes: List[str],
months: List[int] = None, from_year: int = None, to_year: int = None,
any_tag: bool = False) -> dict:
"""
@ -1615,7 +1615,7 @@ async def search_file_suggestion_count(search_str: str, tag_names: List[str], mi
'''
logger.debug(f'search_file_suggestion_count: {stmt} {params}')
with get_db_curs() as curs:
with session_curs(session) as curs:
if deep_match:
curs.execute(stmt, {**params, 'fts_match': deep_match})
deep_estimate = int(curs.fetchone()['estimate'])
@ -2078,7 +2078,7 @@ class BulkTagPreview:
)
def get_bulk_tag_preview(paths: List[str]) -> BulkTagPreview:
def get_bulk_tag_preview(session: Session, paths: List[str]) -> BulkTagPreview:
"""Get preview information for bulk tagging operation.
Returns the count of files that will be tagged and the tags that are shared by ALL files.
@ -2113,34 +2113,33 @@ def get_bulk_tag_preview(paths: List[str]) -> BulkTagPreview:
# First, get all FileGroups and their tags
# We query by stem prefix because unique_files may contain non-primary files
# (e.g., .readability.json) but FileGroups are stored with primary_path (e.g., .html)
with get_db_session() as session:
from sqlalchemy import or_
stems = [split_path_stem_and_suffix(f, full=True)[0] for f in unique_files]
# Chunk the OR'd LIKEs to stay within SQLite's expression-tree depth limit of 1000.
file_groups = []
for idx in range(0, len(stems), 500):
chunk = stems[idx:idx + 500]
file_groups.extend(session.query(FileGroup).filter(
or_(*[FileGroup.primary_path.like(f'{stem}%') for stem in chunk])
).all())
from sqlalchemy import or_
stems = [split_path_stem_and_suffix(f, full=True)[0] for f in unique_files]
# Chunk the OR'd LIKEs to stay within SQLite's expression-tree depth limit of 1000.
file_groups = []
for idx in range(0, len(stems), 500):
chunk = stems[idx:idx + 500]
file_groups.extend(session.query(FileGroup).filter(
or_(*[FileGroup.primary_path.like(f'{stem}%') for stem in chunk])
).all())
if not file_groups:
# No FileGroups exist yet, no shared tags
return BulkTagPreview(file_count=file_count, shared_tag_names=[])
if not file_groups:
# No FileGroups exist yet, no shared tags
return BulkTagPreview(file_count=file_count, shared_tag_names=[])
# Get tag names for each FileGroup
tag_sets = []
for fg in file_groups:
tag_sets.append(set(fg.tag_names))
# Get tag names for each FileGroup
tag_sets = []
for fg in file_groups:
tag_sets.append(set(fg.tag_names))
# Find intersection of all tag sets (tags shared by ALL files)
if tag_sets:
shared_tags = tag_sets[0]
for tag_set in tag_sets[1:]:
shared_tags = shared_tags.intersection(tag_set)
shared_tag_names = sorted(list(shared_tags))
else:
shared_tag_names = []
# Find intersection of all tag sets (tags shared by ALL files)
if tag_sets:
shared_tags = tag_sets[0]
for tag_set in tag_sets[1:]:
shared_tags = shared_tags.intersection(tag_set)
shared_tag_names = sorted(list(shared_tags))
else:
shared_tag_names = []
return BulkTagPreview(file_count=file_count, shared_tag_names=shared_tag_names)

48
wrolpi/files/test/test_lib.py
View file

@ -750,19 +750,19 @@ async def test_files_indexer(async_client, test_session, make_files_structure, t
assert video_file.mimetype == 'video/mp4' and video_file.indexer == indexers.DefaultIndexer
# File are indexed by their titles and contents. Contents (d_text) require a deep search.
files, total = lib.search_files('file', 10, 0)
files, total = lib.search_files(test_session, 'file', 10, 0)
assert total == 5, 'All files contain "file" in their file name. The associated video file is hidden.'
files, total = lib.search_files('image', 10, 0)
files, total = lib.search_files(test_session, 'image', 10, 0)
assert total == 1 and files[0]['title'] == 'an image file.jpeg', 'The image file title contains "image".'
files, total = lib.search_files('contents', 10, 0, deep=True)
files, total = lib.search_files(test_session, 'contents', 10, 0, deep=True)
assert total == 1 and files[0]['title'] == 'a text file.txt', 'The text file contains "contents".'
files, total = lib.search_files('contents', 10, 0)
files, total = lib.search_files(test_session, 'contents', 10, 0)
assert total == 0, 'File contents are not searched unless deep=True.'
files, total = lib.search_files('video', 10, 0)
files, total = lib.search_files(test_session, 'video', 10, 0)
assert total == 1 and {i['title'] for i in files} == {'a video file'}, 'The video file contains "video".'
files, total = lib.search_files('yawn', 10, 0, deep=True)
files, total = lib.search_files(test_session, 'yawn', 10, 0, deep=True)
assert total == 1 and files[0]['title'] == 'a video file', 'The video file captions contain "yawn".'
files, total = lib.search_files('bunny', 10, 0, deep=True)
files, total = lib.search_files(test_session, 'bunny', 10, 0, deep=True)
assert total == 1 and {i['title'] for i in files} == {'a zip file.zip'}, \
'The zip file contains a file with "bunny" in the title.'
@ -774,7 +774,7 @@ async def test_files_indexer(async_client, test_session, make_files_structure, t
text_path.write_text('new text contents')
await refresh_files()
await await_background_tasks()
files, total = lib.search_files('new', 10, 0, deep=True)
files, total = lib.search_files(test_session, 'new', 10, 0, deep=True)
assert total == 1
@ -1238,29 +1238,29 @@ async def test_file_search_date_range(async_client, test_session, example_pdf, e
await refresh_files()
doc, pdf = test_session.query(FileGroup).order_by(FileGroup.primary_path).all()
files, total = lib.search_files('', 10, 0)
files, total = lib.search_files(test_session, '', 10, 0)
assert total == 2, 'Should only be 2 files.'
# Past the last page still reports the true total (count is a separate query).
files, total = lib.search_files('', 10, 10)
files, total = lib.search_files(test_session, '', 10, 10)
assert files == []
assert total == 2
# PDF was published in December.
files, total = lib.search_files('', 10, 0, months=[12, ])
files, total = lib.search_files(test_session, '', 10, 0, months=[12, ])
assert total == 1, 'Only PDF should be from December 2022'
assert files[0]['id'] == pdf.id
# PDF was published in 2022.
files, total = lib.search_files('', 10, 0, from_year=2022, to_year=2022)
files, total = lib.search_files(test_session, '', 10, 0, from_year=2022, to_year=2022)
assert total == 1, 'Only PDF should be from December 2022'
assert files[0]['id'] == pdf.id
# PDF was NOT published in 2023.
files, total = lib.search_files('', 10, 0, from_year=2023)
files, total = lib.search_files(test_session, '', 10, 0, from_year=2023)
assert total == 0, 'No example files are published in 2023'
files, total = lib.search_files('', 10, 0, to_year=2022)
files, total = lib.search_files(test_session, '', 10, 0, to_year=2022)
assert total == 1
assert files[0]['id'] == pdf.id
@ -1358,7 +1358,7 @@ async def test_get_bulk_tag_preview_files(async_client, test_session, make_files
test_session.commit()
# Preview for all three files
preview = lib.get_bulk_tag_preview(['foo.txt', 'bar.txt', 'baz.txt'])
preview = lib.get_bulk_tag_preview(test_session, ['foo.txt', 'bar.txt', 'baz.txt'])
assert preview.file_count == 3
assert 'shared' in preview.shared_tag_names
assert 'only_foo' not in preview.shared_tag_names # Not shared by all
@ -1374,7 +1374,7 @@ async def test_get_bulk_tag_preview_directory(test_session, make_files_structure
})
# Preview for the directory - should find all files recursively
preview = lib.get_bulk_tag_preview(['mydir/'])
preview = lib.get_bulk_tag_preview(test_session, ['mydir/'])
assert preview.file_count == 3
@ -1426,7 +1426,7 @@ async def test_get_bulk_tag_preview_many_files(test_session, make_files_structur
stems exceeded SQLite's expression-tree depth limit ("Expression tree is too large")."""
make_files_structure({f'mydir/file{i}.txt': str(i) for i in range(1200)})
preview = lib.get_bulk_tag_preview(['mydir/'])
preview = lib.get_bulk_tag_preview(test_session, ['mydir/'])
assert preview.file_count == 1200
@ -1459,14 +1459,14 @@ async def test_get_bulk_tag_preview_multi_file_filegroup(async_client, test_sess
test_session.commit()
# Preview for the directory should find the FileGroup and its tag
preview = lib.get_bulk_tag_preview(['archive/'])
preview = lib.get_bulk_tag_preview(test_session, ['archive/'])
assert preview.file_count == 1, f'Expected 1 FileGroup, got {preview.file_count}'
assert 'my_tag' in preview.shared_tag_names, f'Expected my_tag in shared tags, got {preview.shared_tag_names}'
def test_get_bulk_tag_preview_empty(test_session, test_directory):
"""get_bulk_tag_preview returns 0 for empty or non-existent paths."""
preview = lib.get_bulk_tag_preview([])
preview = lib.get_bulk_tag_preview(test_session, [])
assert preview.file_count == 0
assert preview.shared_tag_names == []
@ -1758,21 +1758,21 @@ async def test_file_search_url(test_session, make_files_structure, refresh_files
test_session.commit()
# Search by full URL
files, total = lib.search_files('', 10, 0, url='https://example.com/foo')
files, total = lib.search_files(test_session, '', 10, 0, url='https://example.com/foo')
assert total == 1
assert files[0]['id'] == foo_fg.id
# Search by domain (partial match)
files, total = lib.search_files('', 10, 0, url='example.com')
files, total = lib.search_files(test_session, '', 10, 0, url='example.com')
assert total == 2
assert {f['id'] for f in files} == {foo_fg.id, baz_fg.id}
# Case insensitive search
files, total = lib.search_files('', 10, 0, url='EXAMPLE.COM')
files, total = lib.search_files(test_session, '', 10, 0, url='EXAMPLE.COM')
assert total == 2
# No results for non-matching URL
files, total = lib.search_files('', 10, 0, url='nonexistent.com')
files, total = lib.search_files(test_session, '', 10, 0, url='nonexistent.com')
assert total == 0
@ -1790,7 +1790,7 @@ async def test_file_search_url_with_null(test_session, make_files_structure, ref
# bar_fg.url remains None
test_session.commit()
files, total = lib.search_files('', 10, 0, url='example.com')
files, total = lib.search_files(test_session, '', 10, 0, url='example.com')
assert total == 1
assert files[0]['id'] == foo_fg.id

35
wrolpi/root_api.py
View file

@ -439,8 +439,8 @@ async def restart_download(request: Request, download_id: int):
@api_bp.get('/download')
@openapi.description('Get all Downloads so they can be displayed to the User.')
async def get_downloads(_: Request):
data = download_manager.get_fe_downloads()
async def get_downloads(request: Request):
data = download_manager.get_fe_downloads(request.ctx.session)
# Convert `destination` to relative.
for download in data['once_downloads']:
@ -455,8 +455,9 @@ async def get_downloads(_: Request):
@api_bp.post('/download/<download_id:int>/kill')
@openapi.description('Kill a download. It will be stopped if it is pending.')
async def kill_download(_: Request, download_id: int):
download_manager.kill_download(download_id)
async def kill_download(request: Request, download_id: int):
with request.ctx.db.write() as session:
download_manager.kill_download(session, download_id)
return response.empty()
@ -493,8 +494,9 @@ async def clear_failed(request: Request):
@api_bp.post('/download/retry_once')
@openapi.description('Retry failed once-downloads')
async def retry_once(_: Request):
download_manager.retry_downloads(reset_attempts=True)
async def retry_once(request: Request):
with request.ctx.db.write() as session:
download_manager.retry_downloads(session, reset_attempts=True)
return response.empty()
@ -563,7 +565,7 @@ async def get_status(request: Request):
downloads = dict()
if flags.db_up.is_set():
try:
downloads = download_manager.get_summary()
downloads = download_manager.get_summary(request.ctx.session)
except Exception as e:
logger.debug('Unable to get download status', exc_info=e)
@ -605,9 +607,9 @@ async def get_status(request: Request):
@openapi.definition(
description='Get summary statistics of all files',
)
async def get_statistics(_):
async def get_statistics(request: Request):
# Full-table aggregates take many seconds cold on a large library; keep them off the event loop.
file_statistics = await asyncio.to_thread(get_file_statistics)
file_statistics = await asyncio.to_thread(get_file_statistics, request.ctx.session)
global_statistics = await asyncio.to_thread(get_global_statistics)
return json_response({
'file_statistics': file_statistics,
@ -634,13 +636,13 @@ async def feed(request: Request, query: schema.EventsRequest):
description='Get a list of all Tags',
)
async def get_tags_request(request: Request):
tags_ = tags.get_tags()
tags_ = tags.get_tags(request.ctx.session)
tag_names = request.args.getlist('tag_names')
result = dict(tags=tags_)
if tag_names:
limit = request.args.get('limit')
limit = int(limit) if limit is not None else None
result['overlapping_tag_names'] = tags.get_overlapping_tags(tag_names, limit=limit)
result['overlapping_tag_names'] = tags.get_overlapping_tags(request.ctx.session, tag_names, limit=limit)
return json_response(result)
@ -648,8 +650,8 @@ async def get_tags_request(request: Request):
@openapi.definition(
description='Get the most recently used tags',
)
async def get_recent_tags_request(_: Request):
tag_names = tags.get_recent_tags()
async def get_recent_tags_request(request: Request):
tag_names = tags.get_recent_tags(request.ctx.session)
return json_response(dict(tag_names=tag_names))
@ -801,9 +803,10 @@ async def post_search_suggestions(request: Request, body: schema.SearchSuggestio
body=schema.SearchFileEstimateRequest,
validate=True,
)
async def post_search_file_estimates(_: Request, body: schema.SearchFileEstimateRequest):
async def post_search_file_estimates(request: Request, body: schema.SearchFileEstimateRequest):
"""Used by the Global search to suggest FileGroup count to the user."""
counts = await search_file_suggestion_count(
request.ctx.session,
body.search_str,
body.tag_names,
body.mimetypes,
@ -826,9 +829,9 @@ async def post_search_file_estimates(_: Request, body: schema.SearchFileEstimate
body=schema.SearchOtherEstimateRequest,
validate=True,
)
async def post_search_other_estimates(_: Request, body: schema.SearchOtherEstimateRequest):
async def post_search_other_estimates(request: Request, body: schema.SearchOtherEstimateRequest):
"""Used by the Global search to suggest FileGroup count to the user."""
others = await search_other_estimates(body.tag_names)
others = await search_other_estimates(request.ctx.session, body.tag_names)
ret = dict(others=others)
return json_response(ret)

14
wrolpi/tags.py
View file

@ -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_curs, get_db_session, named_placeholders
from wrolpi.db import get_db_session, named_placeholders, session_curs
from wrolpi.downloader import save_downloads_config
from wrolpi.errors import UnknownTag, UsedTag, InvalidTag, FileWorkerConflict, NoPrimaryFile
from wrolpi.events import Events
@ -844,8 +844,8 @@ def sync_tags_directory():
raise
def get_tags() -> List[dict]:
with get_db_curs() as curs:
def get_tags(session: Session) -> List[dict]:
with session_curs(session) as curs:
curs.execute('''
SELECT t.id,
t.name,
@ -862,8 +862,8 @@ def get_tags() -> List[dict]:
return tags
def get_recent_tags(limit: int = 5) -> List[str]:
with get_db_curs() as curs:
def get_recent_tags(session: Session, limit: int = 5) -> List[str]:
with session_curs(session) as curs:
curs.execute('''
SELECT t.name
FROM (SELECT tag_id, MAX(created_at) AS latest
@ -881,7 +881,7 @@ def get_recent_tags(limit: int = 5) -> List[str]:
return [row['name'] for row in curs.fetchall()]
def get_overlapping_tags(tag_names: Union[str, List[str]], limit: int = 5) -> List[str]:
def get_overlapping_tags(session: Session, tag_names: Union[str, List[str]], limit: int = 5) -> List[str]:
if isinstance(tag_names, str):
tag_names = [tag_names]
@ -890,7 +890,7 @@ def get_overlapping_tags(tag_names: Union[str, List[str]], limit: int = 5) -> Li
params = dict(tag_count=len(tag_names), limit=limit)
tag_names_placeholders = named_placeholders('tag_name', tag_names, params)
with get_db_curs() as curs:
with session_curs(session) as curs:
curs.execute(f'''
WITH target_tags AS (SELECT id FROM tag WHERE name IN ({tag_names_placeholders})),
file_matching AS (

2
wrolpi/test/test_api_fixtures.py
View file

@ -203,7 +203,7 @@ async def test_downloads_summary_fixture_matches_the_download_manager(async_clie
"""
from wrolpi.downloader import download_manager
declared = set(download_manager.get_summary())
declared = set(download_manager.get_summary(test_session))
source = TEST_FIXTURES_JS.read_text()
block = re.search(r'^ downloads: \{(.*?)\n },', source, re.S | re.M)

15
wrolpi/test/test_downloader.py
View file

@ -1265,7 +1265,8 @@ async def test_retry_downloads_includes_failed(test_session, test_download_manag
d4.complete() # completed download should not be retried
test_session.commit()
test_download_manager.retry_downloads(reset_attempts=True)
test_download_manager.retry_downloads(test_session, reset_attempts=True)
test_session.commit()
test_session.refresh(d1)
test_session.refresh(d2)
@ -2016,26 +2017,26 @@ async def test_summary_daily_limit_reached(test_session, test_download_manager,
name = test_downloader.name
# No limits configured.
assert test_download_manager.get_summary()['daily_limit_reached'] is False
assert test_download_manager.get_summary(test_session)['daily_limit_reached'] is False
config.download_daily_limit_per_domain = 2
# No due downloads => not reached, even though limits are configured.
_make_processed_download(test_session, name, 'https://example.com/1')
_make_processed_download(test_session, name, 'https://example.com/2')
test_session.commit()
assert test_download_manager.get_summary()['daily_limit_reached'] is False
assert test_download_manager.get_summary(test_session)['daily_limit_reached'] is False
# A due download for the capped domain => reached.
blocked = Download(url='https://example.com/new', downloader=name, status='new')
test_session.add(blocked)
test_session.commit()
assert test_download_manager.get_summary()['daily_limit_reached'] is True
assert test_download_manager.get_summary(test_session)['daily_limit_reached'] is True
# A due download for another domain can still dispatch => not reached.
allowed = Download(url='https://rumble.com/new', downloader=name, status='new')
test_session.add(allowed)
test_session.commit()
assert test_download_manager.get_summary()['daily_limit_reached'] is False
assert test_download_manager.get_summary(test_session)['daily_limit_reached'] is False
# Recurring downloads are never gated by daily limits and must not count as "due".
test_session.delete(allowed)
@ -2043,7 +2044,7 @@ async def test_summary_daily_limit_reached(test_session, test_download_manager,
frequency=DownloadFrequency.daily)
test_session.add(recurring)
test_session.commit()
assert test_download_manager.get_summary()['daily_limit_reached'] is True
assert test_download_manager.get_summary(test_session)['daily_limit_reached'] is True
# Global limit blocks everything, even a fresh domain.
config.download_daily_limit_per_domain = None
@ -2051,7 +2052,7 @@ async def test_summary_daily_limit_reached(test_session, test_download_manager,
fresh = Download(url='https://wikipedia.org/new', downloader=name, status='new')
test_session.add(fresh)
test_session.commit()
assert test_download_manager.get_summary()['daily_limit_reached'] is True
assert test_download_manager.get_summary(test_session)['daily_limit_reached'] is True
@pytest.mark.asyncio

139
wrolpi/test/test_request_db.py
View file

@ -7,13 +7,9 @@ from http import HTTPStatus
import pytest
from wrolpi import flags
from wrolpi.conftest import count_request_connections
# Library functions these endpoints call still open their own cursor or session, instead of using the
# request's. Remove an endpoint's mark when its count reaches one (strict: an unexpected pass fails).
OPENS_OWN_CONNECTION = pytest.mark.xfail(
strict=True, reason='a library function opens its own connection instead of using the request session')
@pytest.mark.asyncio
@pytest.mark.parametrize('method,path,body', [
@ -61,9 +57,19 @@ async def test_create_download_uses_one_connection(async_client, test_session, t
@pytest.mark.asyncio
async def test_tags_use_one_connection(async_client, test_session, tag_factory):
await tag_factory()
response, count = await count_request_connections(async_client, test_session, 'get', '/api/tag')
@pytest.mark.parametrize('path', ['/api/tag', '/api/tag?tag_names={name}', '/api/tag/recent'])
async def test_tags_use_one_connection(async_client, test_session, tag_factory, path):
"""The Tags, their overlaps, and the recent Tags are read on the request's connection."""
tag = await tag_factory()
response, count = await count_request_connections(
async_client, test_session, 'get', path.format(name=tag.name))
assert response.status_code == HTTPStatus.OK, response.body
assert count == 1
@pytest.mark.asyncio
async def test_channel_list_uses_one_connection(async_client, test_session, simple_channel):
response, count = await count_request_connections(async_client, test_session, 'get', '/api/videos/channels')
assert response.status_code == HTTPStatus.OK, response.body
assert count == 1
@ -77,10 +83,9 @@ async def test_zim_estimates_use_one_connection(async_client, test_session, zim_
assert count == 1
@OPENS_OWN_CONNECTION
@pytest.mark.asyncio
async def test_zim_search_uses_one_connection(async_client, test_session, zim_factory):
"""`get_all_entries_tags` opens a cursor of its own."""
"""The zim's entries' tags are read on the request's connection."""
zim_factory('test zim')
response, count = await count_request_connections(
async_client, test_session, 'post', '/api/zim/search', json={'search_str': 'one'})
@ -88,25 +93,119 @@ async def test_zim_search_uses_one_connection(async_client, test_session, zim_fa
assert count == 1
@OPENS_OWN_CONNECTION
@pytest.mark.asyncio
@pytest.mark.parametrize('path', ['/api/files/search', '/api/videos/search'])
async def test_search_uses_one_connection(async_client, test_session, simple_channel, video_factory, path):
"""`count_file_groups` opens a cursor for the total (skipped when the total is cached), then
`handle_file_group_search_results` opens another cursor and a separate session for the page."""
@pytest.mark.parametrize('path,body', [
('/api/files/search', {'search_str': 'video'}),
('/api/videos/search', {'search_str': 'video'}),
('/api/archive/search', {'search_str': 'video'}),
('/api/flasher/search', {}),
])
async def test_search_uses_one_connection(async_client, test_session, simple_channel, video_factory, path, body):
"""The search total, the page of ids, and the models for that page share the request's connection."""
video_factory(title='a video', channel_id=simple_channel.id)
response, count = await count_request_connections(
async_client, test_session, 'post', path, json={'search_str': 'video'})
response, count = await count_request_connections(async_client, test_session, 'post', path, json=body)
assert response.status_code == HTTPStatus.OK, response.body
assert count == 1
@OPENS_OWN_CONNECTION
@pytest.mark.asyncio
async def test_channel_uses_one_connection(async_client, test_session, simple_channel):
"""`get_channel` uses the request session, but `Channel.get_statistics` opens a raw cursor of its own
(`get_db_curs`)."""
"""The Channel and its statistics are read on the request's connection."""
response, count = await count_request_connections(
async_client, test_session, 'get', f'/api/videos/channels/{simple_channel.id}')
assert response.status_code == HTTPStatus.OK, response.body
assert count == 1
@pytest.mark.asyncio
async def test_zim_tag_searches_use_one_connection(async_client, test_session, test_zim, tag_factory):
"""Searching a Zim's tagged entries reads the tags on the request's connection."""
tag = await tag_factory()
test_zim.tag_entry(test_session, tag.name, 'one')
response, count = await count_request_connections(
async_client, test_session, 'post', f'/api/zim/search/{test_zim.id}', json={'tag_names': [tag.name]})
assert response.status_code == HTTPStatus.OK, response.body
assert count == 1
response, count = await count_request_connections(
async_client, test_session, 'post', '/api/zim/search_estimates', json={'tag_names': [tag.name]})
assert response.status_code == HTTPStatus.OK, response.body
assert count == 1
@pytest.mark.asyncio
@pytest.mark.parametrize('suffix', ['', '/description', '/captions', '/comments'])
async def test_video_reads_use_one_connection(async_client, test_session, simple_channel, video_factory, suffix):
"""A Video, its neighbors, and its details are read on the request's connection."""
video_factory(title='first', channel_id=simple_channel.id)
video = video_factory(title='second', channel_id=simple_channel.id)
response, count = await count_request_connections(
async_client, test_session, 'get', f'/api/videos/{video.file_group_id}{suffix}')
assert response.status_code == HTTPStatus.OK, response.body
assert count == 1
@pytest.mark.asyncio
async def test_video_edit_uses_one_connection(async_client, test_session, simple_channel, video_factory):
video = video_factory(title='before', channel_id=simple_channel.id)
response, count = await count_request_connections(
async_client, test_session, 'put', f'/api/videos/{video.file_group_id}', json={'title': 'after'})
assert response.status_code == HTTPStatus.OK, response.body
assert response.json['file_group']['title'] == 'after'
assert count == 1
@pytest.mark.asyncio
@pytest.mark.parametrize('method,path,body', [
('get', '/api/statistics', None),
('post', '/api/search_file_estimates', {'search_str': 'video'}),
('post', '/api/search_other_estimates', {'tag_names': ['{tag}']}),
('post', '/api/files/bulk_tag/preview', {'paths': ['{video_dir}']}),
('get', '/api/download', None),
('post', '/api/download/retry_once', None),
('get', '/api/docs/statistics', None),
('get', '/api/videos/statistics', None),
('get', '/api/archive/statistics', None),
('post', '/api/docs/search', {'search_str': 'video'}),
])
async def test_summaries_use_one_connection(async_client, test_session, test_directory, tag_factory, simple_channel,
video_factory, test_download_manager, method, path, body):
"""Statistics, estimates, previews and lists each read on the request's connection."""
tag = await tag_factory()
video = video_factory(title='a video', channel_id=simple_channel.id)
video_dir = str(video.video_path.parent.relative_to(test_directory))
if body is not None:
body = {k: [i.format(tag=tag.name, video_dir=video_dir) for i in v] if isinstance(v, list) else v
for k, v in body.items()}
kwargs = dict(json=body) if body is not None else dict()
response, count = await count_request_connections(async_client, test_session, method, path, **kwargs)
assert response.status_code < 300, response.body
assert count == 1
@pytest.mark.asyncio
async def test_status_uses_one_connection(async_client, test_session, test_download_manager, flags_lock):
"""The UI polls the status; its download summary is read on the request's connection."""
flags.db_up.set()
try:
response, count = await count_request_connections(async_client, test_session, 'get', '/api/status')
finally:
flags.db_up.clear()
assert response.status_code == HTTPStatus.OK, response.body
assert 'pending' in response.json['downloads'], 'the download summary was not read'
assert count == 1
@pytest.mark.asyncio
async def test_kill_download_uses_one_connection(async_client, test_session, test_downloader):
from wrolpi.downloader import Download
await async_client.post('/api/download', json={'urls': ['https://example.com/kill'],
'downloader': test_downloader.name})
download = test_session.query(Download).filter_by(url='https://example.com/kill').one()
response, count = await count_request_connections(
async_client, test_session, 'post', f'/api/download/{download.id}/kill')
assert response.status_code < 300, response.body
assert count == 1
assert test_session.query(Download).filter_by(id=download.id).one().error == 'User stopped this download'

16
wrolpi/test/test_tags.py
View file

@ -730,11 +730,11 @@ async def test_get_recent_tags(test_session, make_files_structure, tag_factory,
tf3.created_at = base
test_session.commit()
result = tags.get_recent_tags(limit=5)
result = tags.get_recent_tags(test_session, limit=5)
assert result == ['gamma', 'beta', 'alpha']
# Limit is respected.
result = tags.get_recent_tags(limit=2)
result = tags.get_recent_tags(test_session, limit=2)
assert result == ['gamma', 'beta']
@ -757,7 +757,7 @@ async def test_get_overlapping_tags(test_session, make_files_structure, tag_fact
file_groups[1].add_tag(test_session, tag2.id)
test_session.commit()
result = tags.get_overlapping_tags('alpha')
result = tags.get_overlapping_tags(test_session, 'alpha')
# beta overlaps with alpha twice, gamma once.
assert result == ['beta', 'gamma']
@ -766,7 +766,7 @@ async def test_get_overlapping_tags(test_session, make_files_structure, tag_fact
async def test_get_overlapping_tags_empty(test_session, tag_factory):
"""No overlaps returns an empty list."""
await tag_factory('lonely')
result = tags.get_overlapping_tags('lonely')
result = tags.get_overlapping_tags(test_session, 'lonely')
assert result == []
@ -814,11 +814,11 @@ async def test_get_overlapping_tags_multiple(test_session, make_files_structure,
test_session.commit()
# Overlapping with both alpha AND beta: gamma (file a), delta (file b).
result = tags.get_overlapping_tags(['alpha', 'beta'])
result = tags.get_overlapping_tags(test_session, ['alpha', 'beta'])
assert set(result) == {'gamma', 'delta'}
# Overlapping with both alpha AND gamma: beta (file a only, since file c has no beta).
result = tags.get_overlapping_tags(['alpha', 'gamma'])
result = tags.get_overlapping_tags(test_session, ['alpha', 'gamma'])
assert 'beta' in result
@ -842,11 +842,11 @@ async def test_get_overlapping_tags_no_limit(test_session, make_files_structure,
test_session.commit()
# Default limit of 5 returns only 5.
result = tags.get_overlapping_tags('alpha')
result = tags.get_overlapping_tags(test_session, 'alpha')
assert len(result) == 5
# No limit returns all 10.
result = tags.get_overlapping_tags('alpha', limit=None)
result = tags.get_overlapping_tags(test_session, 'alpha', limit=None)
assert len(result) == 10