From 6434d5ca20cdea75893d374d10cd1dad9ef606cc Mon Sep 17 00:00:00 2001 From: lrnselfreliance <33263708+lrnselfreliance@users.noreply.github.com> Date: Sat, 24 Oct 2026 18:49:15 -0600 Subject: [PATCH] Database configs import by diffing; download_manager.yaml first (#867) * Database configs import by diffing; download_manager.yaml first Step 2 of the config import rework. A database config can now declare a ConfigSection for each kind of row it saves, and ConfigFile.import_config runs a fixed sequence for it: - read and validate the file once; - decide from the versions (wrolpi.config_version) what the import may do; - snapshot each section's rows as plain data keyed the way the file names them, and diff: added, changed, removed (none from an empty section, or without a current version); - prepare (files, network) with no transaction open; - write only the differences, and the file's version, in one short write. A failure writes nothing. If the database's version changed since the snapshot, the import decides again (and saves a newer database). Subclasses override the section methods (file_rows, db_rows, upsert, delete) and the prepare/after_import hooks. The other database configs keep their importers until their own steps. download_manager.yaml is the first: an unchanged config imports with no write (was 5,000 statements for 2,500 Downloads), new Downloads are one INSERT, and a missing ZimSubscription is restored without rewriting the rest. Settings equal to the global defaults compare as unset, as they are saved. A destination written relative to the media directory is stored absolute. Completed once-downloads are never removed (they are not saved; #679). Co-Authored-By: Claude Opus 5.5 * Sync Zim subscriptions directly during the downloads import A row now compares its subscription's (name, language), derived from the URL, instead of whether one exists. The import moves a subscription of that name onto the written Download (a changed language or URL), removes one the file no longer implies, and leaves the Download's own columns as the file has them; change_download renamed the old Download's URL into the one just inserted, and reset attempts and the downloaders. A Kiwix URL that names no known subscription is a plain Download instead of failing the whole write. prepare runs only when there are differences, and its docstring states that it must be safe to repeat and to abandon. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- wrolpi/common.py | 125 ++++++++++ wrolpi/config_import.py | 73 ++++++ wrolpi/downloader.py | 306 +++++++++++++---------- wrolpi/test/test_config_import.py | 386 ++++++++++++++++++++++++++++++ wrolpi/test/test_downloader.py | 6 +- 5 files changed, 761 insertions(+), 135 deletions(-) create mode 100644 wrolpi/config_import.py create mode 100644 wrolpi/test/test_config_import.py diff --git a/wrolpi/common.py b/wrolpi/common.py index c04d60ff..bcefab5a 100644 --- a/wrolpi/common.py +++ b/wrolpi/common.py @@ -494,6 +494,10 @@ DB_CONFIG_FILE_NAMES ={'tags.yaml', 'channels.yaml', 'domains.yaml', 'download_m 'playlists.yaml', 'repos.yaml'} +class _DatabaseChanged(Exception): + """The database changed a config's rows between an import's snapshot and its write.""" + + class ConfigFile: """ This is used to keep the DB and config files in sync. Importing the config reads the config from file, then @@ -759,8 +763,15 @@ class ConfigFile: def get_relative_file(self): return get_relative_to_media_directory(self.get_file()) + # A database config that imports by diffing (see wrolpi.config_import) lists a `ConfigSection` for each kind of + # row it saves, parents first. Its `import_config` is the fixed sequence in `import_sections`. + sections: tuple = () + def import_config(self, file: pathlib.Path = None, send_events=False): """Read config file data, apply it to the in-memory config and database.""" + if self.sections: + return self.import_sections(file, send_events) + file = file or self.get_file() file_str = str(get_relative_to_media_directory(file)) # Caller will set to successful if it works. @@ -782,6 +793,109 @@ class ConfigFile: else: logger_.error(f'Failed to import {file_str} because it does not exist.') + def import_sections(self, file: pathlib.Path = None, send_events=False): + """Import a config with `sections`: read and validate the file once, decide what its version allows (see + wrolpi.config_version), diff each section against a snapshot of the database, prepare with no transaction + open, then write only the differences, and the file's version, in one short write.""" + from wrolpi.events import Events + file = file or self.get_file() + file_str = str(get_relative_to_media_directory(file)) + forced, self.force_import = self.force_import, False + self.successful_import = False + self.import_skipped = 0 + try: + if not file.is_file(): + # A missing config is not an empty one: import nothing, delete nothing. + logger_.info(f'No config file, skipping import: {file_str}') + self.successful_import = True + return + + data = self.read_config_file(file) if file.stat().st_size else None + if not data or not self.validate(data): + raise InvalidConfig(f'Config is {"invalid" if data else "empty"}: {file_str}') + self._config.update({k: v for k, v in data.items() if k in self.default_config}) + + # Another writer may change the database between the snapshot and the write; then decide again. + for _ in range(3): + if self._import_sections_once(forced): + break + else: + raise RuntimeError(f'The database kept changing while importing {file_str}') + self.successful_import = True + except Exception as e: + self.successful_import = False + message = f'Failed to import {file_str}' + logger_.error(message, exc_info=e) + if send_events: + Events.send_config_import_failed(message) + raise + + def _import_sections_once(self, forced: bool) -> bool: + """One pass of `import_sections`. Returns False when the database's version changed before the write.""" + from wrolpi.config_import import diff_section + from wrolpi.config_version import IMPORT, SAVE, get_config_version, set_config_version + from wrolpi.db import get_db_session + + decision, db_version = self.version_decision(forced) + if decision == SAVE: + self.save_database_instead() + return True + + file_rows = {i.name: i.file_rows(self._config) for i in self.sections} + with get_db_session() as session: + db_rows = {i.name: i.db_rows(session) for i in self.sections} + diffs = {i.name: diff_section(*file_rows[i.name], db_rows[i.name], deletes=decision == IMPORT) + for i in self.sections} + if any(diffs.values()): + self.prepare(diffs) + + file_version = self.version or 0 + if any(diffs.values()) or (decision == IMPORT and db_version != file_version): + try: + with get_db_session(commit=True) as session: + try: + if get_config_version(session, self.file_name) != db_version: + raise _DatabaseChanged() + for section in self.sections: + if rows := {**diffs[section.name].added, **diffs[section.name].changed}: + section.upsert(session, rows) + for section in reversed(self.sections): + if diffs[section.name].removed: + section.delete(session, diffs[section.name].removed) + if decision == IMPORT: + # After the writes above, which counted as changes. + session.flush() + set_config_version(session, self.file_name, file_version) + except BaseException: + # None of it, and release the write lock now. + session.rollback() + raise + except _DatabaseChanged: + logger_.info(f'{self.file_name} changed in the database while importing; deciding again') + return False + + self.import_skipped = sum(len(invalid) for _, invalid in file_rows.values()) + changed = {name: diff for name, diff in diffs.items() if diff} + logger_.info(f'Imported {self.file_name} ({decision}): ' + + (', '.join(f'{name}: {len(diff.added)} added, {len(diff.changed)} changed, ' + f'{len(diff.removed)} removed' for name, diff in changed.items()) or 'no changes')) + self.after_import(diffs) + if decision != IMPORT: + # The database may have rows the file does not list; save them (see `record_import_version`). + self.background_dump.activate_switch() + return True + + def prepare(self, diffs: dict): + """Slow work the write needs (files, network), run with no transaction open, and only when there are + differences. `diffs` maps each section's name to its `SectionDiff`. + + It must be safe to repeat and to abandon: when the database changes before the write, the import decides + again (up to three passes), and the diffs it was given may never be written (the write fails, or the next + pass saves the database instead).""" + + def after_import(self, diffs: dict): + """Run after the write: activate switches, send events, start background tasks.""" + def dump_config(self, file: pathlib.Path = None, send_events=False, overwrite=False): """Dump database data, copy it to this the in-memory config, then write it to the config file.""" # Copy the data as-is by default. Other classes will overwrite this and dump the database before saving. @@ -830,6 +944,17 @@ class ConfigFile: self.last_import_decision = decision return decision + def version_decision(self, forced: bool = False) -> Tuple[str, Optional[int]]: + """What importing this config's file (already read) may do, and the database version that decided it. For + `import_sections`, which records the version in its own write (`import_decision` is for the others).""" + from wrolpi.config_version import IMPORT, decide, get_config_version + from wrolpi.db import get_db_session + with get_db_session() as session: + db_version = get_config_version(session, self.file_name) + decision = IMPORT if forced else decide(self.version or 0, db_version) + logger_.debug(f'{self.file_name}: file version {self.version}, database version {db_version}: {decision}') + return decision, db_version + def save_database_instead(self): """The database has changes this config's file does not (see `import_decision`); save them.""" logger_.warning(f'{self.file_name} is older than the database; saving the database instead of importing it') diff --git a/wrolpi/config_import.py b/wrolpi/config_import.py new file mode 100644 index 00000000..d05798ff --- /dev/null +++ b/wrolpi/config_import.py @@ -0,0 +1,73 @@ +"""Importing a database config: read its file, diff it against a snapshot of the database, then write only the +differences in one short write (see `ConfigFile.import_config`). + +A config declares a `ConfigSection` for each kind of row it saves (tags.yaml has Tags and the files they tag). +Rows are compared as plain data keyed the way the file names them (a URL, a name, a path), so the snapshot needs no +ORM objects and the read can finish before any slow preparation starts. +""" +from dataclasses import dataclass, field +from typing import Any, Dict, Hashable, Iterable, List, Set, Tuple + +from sqlalchemy.orm import Session + + +@dataclass +class SectionDiff: + """What applying a config file changes in one section. Rows are the file's, as `ConfigSection.file_rows` + returns them.""" + added: Dict[Hashable, dict] = field(default_factory=dict) + changed: Dict[Hashable, dict] = field(default_factory=dict) + removed: Set[Hashable] = field(default_factory=set) + + def __bool__(self): + return bool(self.added or self.changed or self.removed) + + +class ConfigSection: + """One kind of row a database config saves. + + Subclasses define how a file row and a database row become the same comparable dict (the values the config + saves, keyed by `key`), and how to write the differences.""" + name: str = None + + def file_rows(self, config: dict) -> Tuple[Dict[Hashable, dict], Set[Hashable]]: + """The rows this section of `config` (the file's data) lists, keyed, plus the keys of rows that are invalid. + An invalid row is skipped, but its key counts as listed, so it is never deleted.""" + raise NotImplementedError() + + def db_rows(self, session: Session) -> Dict[Hashable, dict]: + """The rows the config saves, keyed and compared like `file_rows`. Rows the config never saves are left + out: they are never "removed".""" + raise NotImplementedError() + + def upsert(self, session: Session, rows: Dict[Hashable, dict]): + """Write these added or changed rows. Fetch the current rows by key first: one may have been created or + deleted since the snapshot.""" + raise NotImplementedError() + + def delete(self, session: Session, keys: Set[Hashable]): + """Delete the rows with these keys. A row deleted since the snapshot, or no longer saved by the config, is + left alone.""" + raise NotImplementedError() + + +def diff_section(file_rows: Dict[Hashable, dict], invalid_keys: Set[Hashable], db_rows: Dict[Hashable, dict], + deletes: bool) -> SectionDiff: + """Compare a section's file rows with its database rows. `deletes`: rows the file no longer lists are removed. + An empty section never removes anything: it cannot be told apart from a lost or blanked file.""" + diff = SectionDiff() + for key, row in file_rows.items(): + if key not in db_rows: + diff.added[key] = row + elif db_rows[key] != row: + diff.changed[key] = row + if deletes and (file_rows or invalid_keys): + diff.removed = db_rows.keys() - file_rows.keys() - invalid_keys + return diff + + +def chunks(items: Iterable[Any], size: int = 500) -> Iterable[List[Any]]: + """Split `items` for `IN (...)` queries (SQLite limits the parameters of one statement).""" + items = list(items) + for i in range(0, len(items), size): + yield items[i:i + size] diff --git a/wrolpi/downloader.py b/wrolpi/downloader.py index bd51c21f..90414bbb 100644 --- a/wrolpi/downloader.py +++ b/wrolpi/downloader.py @@ -17,14 +17,14 @@ from enum import Enum from http import HTTPStatus from itertools import filterfalse from typing import List, Dict, Generator, Iterable, Coroutine, Any -from typing import Tuple, Optional, Callable, Awaitable +from typing import Tuple, Optional, Callable, Awaitable, Set from urllib.parse import urlparse from zoneinfo import ZoneInfo import feedparser import pytz from feedparser import FeedParserDict -from sqlalchemy import Column, Integer, String, Text, ForeignKey, Index, JSON, text +from sqlalchemy import Column, Integer, String, Text, ForeignKey, Index, JSON, insert, or_, text from sqlalchemy.exc import OperationalError from sqlalchemy.orm import Mapped, Session, column_property, relationship @@ -34,7 +34,8 @@ from wrolpi.cmd import which, run_command, CommandResult from wrolpi.common import loaded, Base, ModelHelper, logger, wrol_mode_check, zig_zag, ConfigFile, \ 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.config_version import IMPORT, SAVE, collection_kinds, count_config_change, register_ownership +from wrolpi.config_import import ConfigSection, chunks +from wrolpi.config_version import collection_kinds, count_config_change, register_ownership from wrolpi.dates import TZDateTime, now, Seconds from wrolpi.db import get_db_session, session_curs, serializer from wrolpi.errors import InvalidDownload, UnrecoverableDownloadError, BotBlockedDownloadError, UnknownDownload, \ @@ -413,6 +414,13 @@ class Download(ModelHelper, Base): # noqa return download raise UnknownDownload(f'Cannot find Download with id: {id_}') + @staticmethod + def saved_by_config_filter() -> tuple: + """Filters the Downloads download_manager.yaml saves. A completed once-download is not saved (its + `last_download_attempt` counts toward the daily download limits), so it is never removed by an import.""" + return or_(Download.last_successful_download.is_(None), + Download.frequency.isnot(None) & (Download.frequency != 0)), + @staticmethod def get_by_url(session: Session, url: str) -> Optional['Download']: download = session.query(Download).filter(Download.url == url).one_or_none() @@ -2040,8 +2048,167 @@ class DownloadManagerConfigValidator: skip_urls: list[str] = field(default_factory=list) +def inheritable_video_defaults() -> dict: + from modules.videos.lib import _extract_global_inheritable, get_videos_downloader_config + return _extract_global_inheritable(get_videos_downloader_config()) + + +def _utc(value) -> Optional[datetime]: + """A config's datetime (or ISO string), in UTC; one without a timezone is assumed to be UTC.""" + if not value: + return None + if isinstance(value, str): + value = datetime.fromisoformat(value) + return (value if value.tzinfo else value.replace(tzinfo=pytz.utc)).astimezone(pytz.utc) + + +def zim_subscription_for(url: str, frequency: Optional[int]) -> Optional[Tuple[str, str]]: + """The (name, language) of the ZimSubscription a recurring Kiwix Download belongs to. None for any other + Download, including a Kiwix URL that names no known subscription.""" + from modules.zim.lib import zim_download_url_to_name + if not frequency or not url.startswith('https://download.kiwix.org/'): + return None + try: + return zim_download_url_to_name(url) + except Exception as e: + logger.warning(f'Download {url} is not a known Zim subscription: {e}') + return None + + +class DownloadsSection(ConfigSection): + """download_manager.yaml's Downloads, keyed by URL. A recurring Kiwix Download's ZimSubscription is derived + from it: `subscription` compares the subscription's (name, language), so a missing, stale or unwanted one is + corrected.""" + name = 'downloads' + + @staticmethod + def _row(url, destination, downloader, frequency, last_successful_download, next_download, settings, status, + sub_downloader, tag_names, subscription, defaults: dict) -> dict: + return dict( + destination=str(destination) if destination else None, + downloader=downloader, + frequency=frequency or None, + last_successful_download=_utc(last_successful_download), + next_download=_utc(next_download), + # Settings equal to the global defaults are inherited, and not saved. + settings=DownloadManagerConfig._filter_settings_for_yaml(settings, defaults) or None, + status=status or DownloadStatus.new, + sub_downloader=sub_downloader, + tag_names=list(tag_names) if tag_names else None, + subscription=tuple(subscription) if subscription else None, + ) + + def file_rows(self, config: dict) -> Tuple[Dict[str, dict], Set[str]]: + defaults = inheritable_video_defaults() + rows, invalid = dict(), set() + for data in config.get('downloads') or []: + url = data.get('url') + try: + settings = dict(data.get('settings') or {}) + # An old config kept `destination` and `tag_names` in the settings. + destination = data.get('destination') or settings.pop('destination', None) + tag_names = data.get('tag_names') or settings.pop('tag_names', None) + settings.pop('destination', None) + settings.pop('tag_names', None) + # Stored absolute (relative to the media directory), as the database compares it. + destination = get_absolute_media_path(destination) if destination else None + rows[url] = self._row( + url, destination, data['downloader'], data.get('frequency'), + data.get('last_successful_download'), data.get('next_download'), settings, data.get('status'), + data.get('sub_downloader'), tag_names, zim_subscription_for(url, data.get('frequency')), + defaults) + except Exception as e: + logger.error(f'Skipping invalid Download in config: {data}', exc_info=e) + invalid.add(url) + # A subscription's name is unique: the first Download listed for it gets it. + names = set() + for url, row in rows.items(): + if row['subscription'] and row['subscription'][0] in names: + logger.warning(f'Download {url} is a second Download for Zim subscription {row["subscription"][0]}') + row['subscription'] = None + elif row['subscription']: + names.add(row['subscription'][0]) + return rows, invalid + + def db_rows(self, session: Session) -> Dict[str, dict]: + from modules.zim.models import ZimSubscription + defaults = inheritable_video_defaults() + query = session.query( + Download.url, Download.destination, Download.downloader, Download.frequency, + Download.last_successful_download, Download.next_download, Download.settings, Download.status, + Download.sub_downloader, Download.tag_names, ZimSubscription.name, ZimSubscription.language, + ).outerjoin(ZimSubscription, ZimSubscription.download_id == Download.id) \ + .filter(*Download.saved_by_config_filter()) + return {url: self._row(url, *values, subscription=(name, language) if name else None, defaults=defaults) + for url, *values, name, language in query} + + def upsert(self, session: Session, rows: Dict[str, dict]): + existing = dict() + for urls in chunks(rows): + existing.update({i.url: i for i in session.query(Download).filter(Download.url.in_(urls))}) + values = {url: dict( + destination=row['destination'], + downloader=row['downloader'], + frequency=row['frequency'], + last_successful_download=row['last_successful_download'], + next_download=row['next_download'], + settings=row['settings'] or dict(), + status=row['status'], + sub_downloader=row['sub_downloader'], + tag_names=row['tag_names'], + ) for url, row in rows.items()} + if new := [dict(url=url, **i) for url, i in values.items() if url not in existing]: + # One statement for all new Downloads (a rebuilt database has thousands). Not flushed: count it. + session.execute(insert(Download), new) + count_config_change(session, DownloadManagerConfig.file_name) + for url, download in existing.items(): + for name, value in values[url].items(): + setattr(download, name, value) + session.flush() + self._sync_subscriptions(session, rows) + + @staticmethod + def _sync_subscriptions(session: Session, rows: Dict[str, dict]): + """Give each written Download the ZimSubscription its row names, and no other. A subscription of that name + on another Download (the URL changed) is moved to this one. The Download's own columns are the file's.""" + from modules.zim.models import ZimSubscription + ids = dict() + for urls in chunks(rows): + ids.update(session.query(Download.url, Download.id).filter(Download.url.in_(urls))) + by_download = dict() + for download_ids in chunks(ids.values()): + by_download.update({i.download_id: i for i in + session.query(ZimSubscription).filter(ZimSubscription.download_id.in_(download_ids))}) + for url, row in rows.items(): + download_id = ids[url] + current = by_download.get(download_id) + if current is not None and current.download_id != download_id: + current = None # Moved to an earlier row's Download. + if current is not None and (not row['subscription'] or current.name != row['subscription'][0]): + session.delete(current) + session.flush() + current = None + if row['subscription']: + name, language = row['subscription'] + subscription = current or session.query(ZimSubscription).filter_by(name=name).one_or_none() \ + or ZimSubscription(name=name) + subscription.language = language + subscription.download_id = download_id + session.add(subscription) + session.flush() + + def delete(self, session: Session, keys: Set[str]): + for urls in chunks(keys): + # Re-checked: a once-download may have completed since the snapshot; it is no longer saved. + for download in session.query(Download).filter(Download.url.in_(urls), + *Download.saved_by_config_filter()): + logger.warning(f'Deleting Download {download.url} because it is no longer in the config') + session.delete(download) + + class DownloadManagerConfig(ConfigFile): file_name = 'download_manager.yaml' + sections = (DownloadsSection(),) default_config = dict( downloads=[], skip_urls=[], @@ -2066,14 +2233,13 @@ class DownloadManagerConfig(ConfigFile): self.update({'downloads': value}) @staticmethod - def _filter_settings_for_yaml(settings: dict) -> dict: + def _filter_settings_for_yaml(settings: dict, global_defaults: dict = None) -> dict: """Remove inheritable video settings that match current global defaults. Keeps YAML minimal — only overrides are persisted.""" if not settings: return settings - from modules.videos.lib import INHERITABLE_VIDEO_SETTINGS, _extract_global_inheritable, \ - get_videos_downloader_config - global_defaults = _extract_global_inheritable(get_videos_downloader_config()) + from modules.videos.lib import INHERITABLE_VIDEO_SETTINGS + global_defaults = global_defaults if global_defaults is not None else inheritable_video_defaults() filtered = {} for k, v in settings.items(): if k in INHERITABLE_VIDEO_SETTINGS and v == global_defaults.get(k): @@ -2087,10 +2253,7 @@ class DownloadManagerConfig(ConfigFile): version = self.db_version_for_dump(session) downloads: Iterable[Download] = session.query(Download).order_by(Download.url) new_downloads = [] - for download in downloads: - if download.last_successful_download and not download.frequency: - # This once-download has completed, do not save it. - continue + for download in downloads.filter(*Download.saved_by_config_filter()): destination = download.destination new_download = dict( destination=str(destination) if destination else None, @@ -2113,127 +2276,6 @@ class DownloadManagerConfig(ConfigFile): if send_events: Events.send_config_save_failed(message) - def import_config(self, file: pathlib.Path = None, send_events=False): - file = file or self.get_file() - - # If config file doesn't exist, mark import as successful (nothing to import) - if not file.is_file(): - logger.info('No downloads config file, skipping import') - self.successful_import = True - return - - super().import_config(file) - - decision = self.import_decision() - if decision == SAVE: - self.save_database_instead() - return - - downloads_data = self.downloads - # Empty downloads list = never delete DB records - if not downloads_data: - logger.info(f'No downloads in config, preserving existing DB downloads') - self.successful_import = True - return - - with get_db_session(commit=True) as session: - from modules.zim.lib import zim_download_url_to_name - from modules.zim.models import ZimSubscription - - try: - logger.warning('Importing downloads in config') - - downloads_by_url = {i['url']: i for i in downloads_data} - existing_downloads: List[Download] = list(session.query(Download)) - for existing in existing_downloads: - download = downloads_by_url.pop(existing.url, None) - if download: - # Download in config already exists, update the DB record. - # The config is the source of truth. - existing.downloader = download['downloader'] - existing.destination = download['destination'] - existing.frequency = download['frequency'] - existing.last_successful_download = download.get('last_successful_download') - existing.next_download = download.get('next_download') - existing.status = download['status'] - existing.sub_downloader = download['sub_downloader'] - existing.settings = download.get('settings') or dict() - if 'destination' in existing.settings: - # `destination` may be from old config, or from ChannelDownloader. - destination = existing.settings.pop('destination') - existing.destination = existing.destination or destination - if 'tag_names' in existing.settings: - # `tag_names` may be from old config, or from ChannelDownloader. - tag_names = existing.settings.pop('tag_names') - existing.tag_names = existing.tag_names or tag_names - existing.flush() - elif existing.is_complete and not existing.frequency: - # Completed once-downloads are never written to the config (see `dump_config`), so their - # absence does not mean the user removed them. Keep them: their `last_download_attempt` - # is what the daily download limits count, and that count must survive a restart. - continue - elif decision == IMPORT: - logger.warning(f'Deleting Download {existing.url} because it is no longer in the config') - session.delete(existing) - - for download in downloads_by_url.values(): - # These downloads are new, import them. - download = Download( - destination=download['destination'], - downloader=download['downloader'], - frequency=download['frequency'], - last_successful_download=download.get('last_successful_download'), - next_download=download.get('next_download'), - settings=download['settings'] or dict(), - status=download['status'], - sub_downloader=download['sub_downloader'], - tag_names=download['tag_names'], - url=(url := download['url']), - ) - session.add(download) - logger.info(f'Adding new download {url}') - - session.commit() - self.successful_import = True - except Exception as e: - self.successful_import = False - message = f'Failed to import {self.file_name}' - logger.error(message, exc_info=e) - if send_events: - Events.send_config_import_failed(message) - raise - - try: - # Claim any Kiwix subscriptions for ZimSubscription(s). - downloads: List[Download] = session.query(Download) - need_commit = False - for download in downloads: - if not download.url.startswith('https://download.kiwix.org/') or not download.frequency: - # ZimSubscription requires download.kiwix.org AND a frequency. - continue - - name, language = zim_download_url_to_name(download.url) - subscription = session.query(ZimSubscription).filter_by(name=name, language=language).one_or_none() - if not subscription: - subscription = ZimSubscription(name=name, language=language) - subscription.change_download(session, download.url, download.frequency) - session.add(subscription) - need_commit = True - - if need_commit: - session.commit() - self.record_import_version() - self.successful_import = True - except Exception as e: - self.successful_import = False - message = 'Failed to restore ZimSubscriptions' - logger.error(message, exc_info=e) - if send_events: - Events.send_config_import_failed(message) - raise - - self.successful_import = True - def preview_backup_import(self, backup_date: str, mode: str) -> dict: backup_file = self._get_backup_file(backup_date) backup_data = self.read_config_file(backup_file) diff --git a/wrolpi/test/test_config_import.py b/wrolpi/test/test_config_import.py new file mode 100644 index 00000000..b2f2a0fe --- /dev/null +++ b/wrolpi/test/test_config_import.py @@ -0,0 +1,386 @@ +"""A database config imports by reading its file, diffing it against a snapshot of the database, and writing only the +differences in one short write (see wrolpi.config_import). download_manager.yaml is the first config to import this +way.""" +import sqlite3 +import threading +from contextlib import contextmanager +from datetime import timedelta + +import pytest +import yaml +from sqlalchemy import event, text +from sqlalchemy.engine import Engine + +from modules.zim.models import ZimSubscription +from wrolpi import conftest +from wrolpi.config_version import get_config_version +from wrolpi.dates import now +from wrolpi.db import get_db_session +from wrolpi.downloader import Download, DownloadManagerConfig, get_download_manager_config, import_downloads_config + +WRITES = ('INSERT', 'UPDATE', 'DELETE', 'BEGIN IMMEDIATE') +KIWIX_URL = 'https://download.kiwix.org/zim/wikisource/wikisource_en_all_maxi_' + + +@contextmanager +def statements(): + """Collect the SQL statements run in this block.""" + collected = [] + + def collect(conn, cursor, statement, *_): + collected.append(statement.lstrip()) + + event.listen(Engine, 'before_cursor_execute', collect) + try: + yield collected + finally: + event.remove(Engine, 'before_cursor_execute', collect) + + +def writes(collected): + return [i for i in collected if i.upper().startswith(WRITES)] + + +def database_path(test_session) -> str: + return test_session.get_bind().url.database + + +@pytest.fixture +def production_like_sessions(test_session, saved_downloads): + """Each `get_db_session` gets its own connection, as in production (after `saved_downloads` is committed).""" + test_session.commit() + with conftest.production_like_sessions(test_session) as maker: + yield maker + + +@pytest.fixture +async def saved_downloads(test_session, test_directory, test_download_manager_config, await_switches): + """Downloads of every shape download_manager.yaml saves, saved to the config (which records its version).""" + destination = test_directory / 'videos/channel' + yesterday = now() - timedelta(days=1) + test_session.add_all([ + Download(url='https://example.com/once', downloader='archive', status='new'), + Download(url='https://example.com/recurring', downloader='video_channel', frequency=86400, + sub_downloader='video', destination=destination, tag_names=['one', 'two'], + settings=dict(video_resolutions=['720p']), last_successful_download=yesterday, + next_download=now() + timedelta(days=1), status='complete'), + Download(url='https://example.com/removed', downloader='rss', frequency=3600, status='new'), + # Not saved (completed once-downloads count toward the daily limit), so never "removed". + Download(url='https://example.com/completed', downloader='archive', status='complete', + last_successful_download=yesterday), + ]) + test_session.commit() + from modules.zim import lib as zim_lib + await zim_lib.subscribe(test_session, 'Wikisource', 'en') + test_session.commit() + + get_download_manager_config().dump_config() + await await_switches() + assert get_config_version(test_session, 'download_manager.yaml') is not None + return test_download_manager_config + + +def edit_config(path, change): + data = yaml.safe_load(path.read_text()) + change(data) + path.write_text(yaml.dump(data)) + + +def urls(test_session): + test_session.expire_all() + return {i.url for i in test_session.query(Download)} + + +@pytest.mark.asyncio +async def test_unchanged_import_writes_nothing(test_session, saved_downloads): + """Importing a config the database already matches opens no write transaction.""" + config = get_download_manager_config() + with statements() as collected: + config.import_config() + + assert config.successful_import is True + assert writes(collected) == [] + # The version, then one snapshot of the Downloads (with their Zim subscriptions). + assert len(collected) <= 4, collected + + +@pytest.mark.asyncio +async def test_import_writes_only_the_differences(test_session, saved_downloads): + """An added Download is inserted, a changed one is updated, a removed one is deleted; the others are untouched. + Completed once-downloads are not saved, so they are never removed.""" + def change(data): + downloads = {i['url']: i for i in data['downloads']} + downloads['https://example.com/recurring']['frequency'] = 604800 + del downloads['https://example.com/removed'] + downloads['https://example.com/added'] = dict(url='https://example.com/added', downloader='archive', + status='new', last_successful_download=None, + next_download=None) + data['downloads'] = list(downloads.values()) + + edit_config(saved_downloads, change) + config = get_download_manager_config() + with statements() as collected: + config.import_config() + + assert config.successful_import is True + assert urls(test_session) == {'https://example.com/once', 'https://example.com/recurring', + 'https://example.com/added', 'https://example.com/completed', KIWIX_URL} + assert Download.get_by_url(test_session, 'https://example.com/recurring').frequency == 604800 + download_writes = [i for i in writes(collected) if ' download' in i.lower()] + assert len(download_writes) == 3, download_writes + + +@pytest.mark.asyncio +async def test_zim_subscription_is_restored(test_session, saved_downloads): + """A Kiwix Download's Zim subscription is derived; a missing one is restored, an existing one is left alone.""" + test_session.query(ZimSubscription).delete() + test_session.commit() + + get_download_manager_config().import_config() + + subscription = test_session.query(ZimSubscription).one() + assert (subscription.name, subscription.language, subscription.download.url) == ('Wikisource', 'en', KIWIX_URL) + + +@pytest.mark.asyncio +async def test_failed_write_changes_nothing(test_session, saved_downloads, production_like_sessions, monkeypatch): + """The import's changes and its version are one write: a failure leaves neither.""" + from wrolpi.downloader import DownloadsSection + edit_config(saved_downloads, lambda data: data['downloads'].append(dict( + url='https://example.com/added', downloader='archive', status='new', last_successful_download=None, + next_download=None))) + edit_config(saved_downloads, lambda data: data.update( + version=data['version'] + 1, + downloads=[i for i in data['downloads'] if i['url'] != 'https://example.com/removed'])) + before = get_config_version(test_session, 'download_manager.yaml') + + def fail(*args, **kwargs): + raise RuntimeError('Write failed') + + monkeypatch.setattr(DownloadsSection, 'delete', fail) + config = get_download_manager_config() + with pytest.raises(RuntimeError): + config.import_config() + + assert config.successful_import is False + assert 'https://example.com/added' not in urls(test_session) + assert 'https://example.com/removed' in urls(test_session) + assert get_config_version(test_session, 'download_manager.yaml') == before + + +@pytest.mark.asyncio +async def test_import_waits_for_a_concurrent_writer(test_session, saved_downloads, production_like_sessions): + """Another connection holding the write lock delays the import's write; it does not fail it.""" + edit_config(saved_downloads, lambda data: data['downloads'].append(dict( + url='https://example.com/added', downloader='archive', status='new', last_successful_download=None, + next_download=None))) + locked, release = threading.Event(), threading.Event() + + def hold_lock(): + conn = sqlite3.connect(database_path(test_session), isolation_level=None) + conn.execute('BEGIN IMMEDIATE') + locked.set() + release.wait(5) + conn.execute('COMMIT') + conn.close() + + thread = threading.Thread(target=hold_lock) + thread.start() + locked.wait(5) + threading.Timer(0.5, release.set).start() + try: + get_download_manager_config().import_config() + finally: + release.set() + thread.join() + + with get_db_session() as session: + assert Download.get_by_url(session, 'https://example.com/added') + + +@pytest.mark.asyncio +async def test_write_lock_is_free_while_preparing(test_session, saved_downloads, production_like_sessions, + monkeypatch): + """Slow preparation (files, network) runs with no transaction open: other writers are not blocked.""" + edit_config(saved_downloads, lambda data: data['downloads'].append(dict( + url='https://example.com/added', downloader='archive', status='new', last_successful_download=None, + next_download=None))) + took_lock = [] + + def prepare(self, diffs): + conn = sqlite3.connect(database_path(test_session), isolation_level=None, timeout=0) + conn.execute('BEGIN IMMEDIATE') + conn.execute('COMMIT') + conn.close() + took_lock.append(True) + + monkeypatch.setattr(DownloadManagerConfig, 'prepare', prepare) + get_download_manager_config().import_config() + + assert took_lock == [True] + + +@pytest.mark.asyncio +async def test_database_change_during_import_is_kept(test_session, saved_downloads, production_like_sessions, + monkeypatch, await_switches): + """A Download added after the import read the database (and before it wrote) is not deleted by the older + file: the import decides again, finds the database newer, and saves it.""" + edit_config(saved_downloads, lambda data: data['downloads'].pop()) + prepared = [] + + def prepare(self, diffs): + if not prepared: + with get_db_session(commit=True) as session: + session.add(Download(url='https://example.com/concurrent', downloader='archive', status='new')) + prepared.append(True) + + monkeypatch.setattr(DownloadManagerConfig, 'prepare', prepare) + config = get_download_manager_config() + config.import_config() + await await_switches() + + assert config.successful_import is True + with get_db_session() as session: + assert Download.get_by_url(session, 'https://example.com/concurrent') + saved = {i['url'] for i in yaml.safe_load(saved_downloads.read_text())['downloads']} + assert 'https://example.com/concurrent' in saved + + +@pytest.mark.asyncio +async def test_unknown_version_imports_without_deletes(test_session, saved_downloads, await_switches): + """Without a database version, a Download the file does not list is kept (and then saved into the file).""" + test_session.execute(text('DELETE FROM config_version')) + test_session.commit() + edit_config(saved_downloads, lambda data: data.update( + downloads=[i for i in data['downloads'] if i['url'] != 'https://example.com/removed'])) + + await import_downloads_config() + await await_switches() + + assert 'https://example.com/removed' in urls(test_session) + saved = {i['url'] for i in yaml.safe_load(saved_downloads.read_text())['downloads']} + assert 'https://example.com/removed' in saved + assert get_config_version(test_session, 'download_manager.yaml') is not None + + +@pytest.mark.asyncio +async def test_import_button(test_session, saved_downloads, async_client): + """The Import button applies the file whatever its version (the user chose it), inside the request's write.""" + test_session.add(Download(url='https://example.com/unsaved', downloader='archive', status='new')) + test_session.commit() # The database is now newer than the file. + + request, response = await async_client.post('/api/config/import', json=dict(file_name='download_manager.yaml')) + + assert response.status_code == 204 + assert 'https://example.com/unsaved' not in urls(test_session) + + +@pytest.mark.asyncio +async def test_relative_destination(test_session, test_directory, saved_downloads): + """A destination written relative to the media directory is stored absolute, and is then in step.""" + edit_config(saved_downloads, lambda data: data['downloads'].append(dict( + url='https://example.com/relative', downloader='archive', destination='archive/example.com', + status='new', last_successful_download=None, next_download=None))) + config = get_download_manager_config() + config.import_config() + test_session.expire_all() + assert Download.get_by_url(test_session, 'https://example.com/relative').destination \ + == test_directory / 'archive/example.com' + + with statements() as collected: + config.import_config() + assert writes(collected) == [] + + +def edit_kiwix(path, **values): + def change(data): + for download in data['downloads']: + if download['url'] == KIWIX_URL: + download.update(values) + + edit_config(path, change) + + +@pytest.mark.asyncio +async def test_zim_subscription_follows_a_language_change(test_session, saved_downloads): + """A Kiwix URL changed to another language moves the existing subscription (its name is unique) to the new + Download, with the new language; the old Download is removed.""" + french = KIWIX_URL.replace('_en_', '_fr_') + edit_kiwix(saved_downloads, url=french) + + config = get_download_manager_config() + config.import_config() + + assert config.successful_import is True + subscription = test_session.query(ZimSubscription).one() + assert (subscription.name, subscription.language, subscription.download.url) == ('Wikisource', 'fr', french) + assert KIWIX_URL not in urls(test_session) + + +@pytest.mark.asyncio +async def test_zim_subscription_download_keeps_its_state(test_session, saved_downloads): + """Changing a subscribed Download in the file writes only the file's values: its attempts are kept, and the + next import is in step.""" + test_session.query(Download).filter_by(url=KIWIX_URL).update(dict(attempts=3)) + test_session.commit() + edit_kiwix(saved_downloads, frequency=604800) + + config = get_download_manager_config() + config.import_config() + + test_session.expire_all() + download = Download.get_by_url(test_session, KIWIX_URL) + assert (download.frequency, download.attempts) == (604800, 3) + with statements() as collected: + config.import_config() + assert writes(collected) == [] + + +@pytest.mark.asyncio +async def test_unsubscribed_download_drops_its_subscription(test_session, saved_downloads): + """A Kiwix Download the file keeps, but no longer as recurring, loses its subscription; the next import is in + step.""" + edit_kiwix(saved_downloads, frequency=None) + + config = get_download_manager_config() + config.import_config() + + assert test_session.query(ZimSubscription).count() == 0 + assert KIWIX_URL in urls(test_session) + with statements() as collected: + config.import_config() + assert writes(collected) == [] + + +@pytest.mark.asyncio +async def test_unknown_kiwix_url_is_imported(test_session, saved_downloads): + """A recurring Kiwix URL that names no known subscription is imported as a plain Download; it does not fail the + other Downloads.""" + unknown = 'https://download.kiwix.org/zim/other/unknown.zim' + edit_config(saved_downloads, lambda data: data['downloads'].extend([ + dict(url=unknown, downloader='kiwix_zim', frequency=604800, status='new', last_successful_download=None, + next_download=None), + dict(url='https://example.com/added', downloader='archive', status='new', last_successful_download=None, + next_download=None), + ])) + + config = get_download_manager_config() + config.import_config() + + assert config.successful_import is True + assert {unknown, 'https://example.com/added'} <= urls(test_session) + assert test_session.query(ZimSubscription).count() == 1 + with statements() as collected: + config.import_config() + assert writes(collected) == [] + + +@pytest.mark.asyncio +async def test_nothing_to_write_skips_prepare(test_session, saved_downloads, monkeypatch): + """An import with no differences does not start any preparation.""" + prepared = [] + monkeypatch.setattr(DownloadManagerConfig, 'prepare', lambda self, diffs: prepared.append(diffs)) + + get_download_manager_config().import_config() + + assert prepared == [] diff --git a/wrolpi/test/test_downloader.py b/wrolpi/test/test_downloader.py index 0f54b6a3..cfce9d55 100644 --- a/wrolpi/test/test_downloader.py +++ b/wrolpi/test/test_downloader.py @@ -1054,7 +1054,7 @@ def test_download_settings_accepts_valid_download_order(order): @pytest.mark.asyncio -async def test_downloads_config(test_session, test_download_manager, test_download_manager_config, +async def test_downloads_config(test_session, test_directory, test_download_manager, test_download_manager_config, test_downloader, assert_downloads, tag_factory, await_switches): # Can import with an empty config. await import_downloads_config() @@ -1128,8 +1128,8 @@ async def test_downloads_config(test_session, test_download_manager, test_downlo frequency=None, next_download=datetime(2000, 1, 2, 0, 0, 0, tzinfo=pytz.UTC), downloader='test_downloader', - # The destination was imported from the config. - destination=pathlib.Path('a different directory'), + # The destination was imported from the config, relative to the media directory. + destination=test_directory / 'a different directory', sub_downloader=None, tag_names=None, # Tags were removed. ),