diff --git a/alembic/versions/2026_10_24_1200_config_version.py b/alembic/versions/2026_10_24_1200_config_version.py new file mode 100644 index 00000000..bf202b50 --- /dev/null +++ b/alembic/versions/2026_10_24_1200_config_version.py @@ -0,0 +1,27 @@ +"""Each database config's version, so an import can tell whether the database or the file is newer +(see wrolpi.config_version). + +Revision ID: 2026_10_24_1200 +Revises: 2026_10_08_1200 +""" +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision = '2026_10_24_1200' +down_revision = '2026_10_08_1200' +branch_labels = None +depends_on = None + + +def upgrade(): + # No rows: a config's next import (without deletes) records its version. + op.create_table( + 'config_version', + sa.Column('name', sa.String(), primary_key=True), + sa.Column('version', sa.Integer(), nullable=False), + ) + + +def downgrade(): + op.drop_table('config_version') diff --git a/modules/archive/lib.py b/modules/archive/lib.py index 6657bfdd..d852829c 100644 --- a/modules/archive/lib.py +++ b/modules/archive/lib.py @@ -27,6 +27,7 @@ from wrolpi.collections import Collection from wrolpi.common import get_media_directory, get_relative_to_media_directory, logger, extract_domain, \ 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, chunks +from wrolpi.config_version import IMPORT, SAVE from wrolpi.dates import now, Seconds from wrolpi.db import get_db_session, get_db_curs, session_curs from wrolpi.errors import UnknownArchive, InvalidOrderBy, InvalidDatetime @@ -104,6 +105,11 @@ class DomainsConfig(ConfigFile): ConfigFile.import_config(self, file, send_events) + decision = self.import_decision() + if decision == SAVE: + self.save_database_instead() + return + collections_data = self._config.get('collections', []) # Empty collections list = never delete DB records @@ -158,7 +164,7 @@ class DomainsConfig(ConfigFile): # create their domain (e.g. one downloaded before this import), and deleting it would delete them. with_archives = {i for i, in session.query(Archive.collection_id).distinct()} all_domain_collections = session.query(Collection).filter_by(kind='domain').all() - for collection in all_domain_collections: + for collection in all_domain_collections if decision == IMPORT else []: if collection.name in imported_domains: continue if collection.id in with_archives: @@ -169,6 +175,7 @@ class DomainsConfig(ConfigFile): session.delete(collection) logger.info(f'Successfully imported {len(imported_domains)} domain collections from {file_str}') + self.record_import_version() self.successful_import = True except Exception as e: @@ -184,6 +191,7 @@ class DomainsConfig(ConfigFile): logger.info('Dumping domain collections to config') with get_db_session() as session: + version = self.db_version_for_dump(session) # Get only domain collections, ordered by name collections = session.query(Collection).filter_by(kind='domain').order_by(Collection.name).all() @@ -195,6 +203,7 @@ class DomainsConfig(ConfigFile): return self._config['collections'] = collections_data + self._config['version'] = version logger.info(f'Dumping {len(collections_data)} domain collections to config') self.save(file, send_events, overwrite) @@ -261,7 +270,8 @@ class DomainsConfig(ConfigFile): config_file.parent.mkdir(parents=True, exist_ok=True) self.write_config_data(merged_data, config_file) - self.import_config(send_events=send_events) + # Importing a backup is the user's choice: apply it whatever the versions. + self.import_forced(send_events=send_events) # Global instance diff --git a/modules/archive/test/test_domains_config.py b/modules/archive/test/test_domains_config.py index efe0ead6..2d4aada3 100644 --- a/modules/archive/test/test_domains_config.py +++ b/modules/archive/test/test_domains_config.py @@ -11,6 +11,7 @@ import yaml from modules.archive.lib import get_domains_config from wrolpi.collections import Collection +from wrolpi.conftest import config_was_imported @pytest.mark.asyncio @@ -205,6 +206,7 @@ class TestDomainsConfigEdgeCases: # Reinitialize and import config.initialize() + config_was_imported(test_session, 'domains.yaml') config.import_config() assert config.successful_import is True diff --git a/modules/repos/lib.py b/modules/repos/lib.py index 39a7ef40..b56ef585 100644 --- a/modules/repos/lib.py +++ b/modules/repos/lib.py @@ -25,6 +25,7 @@ from wrolpi.cmd import GIT_BIN, run_command from wrolpi.collections.models import Collection from wrolpi.common import ConfigFile, get_media_directory, get_relative_to_media_directory, logger, \ escape_file_name, background_task, get_wrolpi_config +from wrolpi.config_version import IMPORT, SAVE, register_ownership from wrolpi.db import get_db_session from wrolpi.downloader import Download, DownloadFrequency, download_manager, save_downloads_config from wrolpi.errors import UnknownDirectory, ValidationError @@ -1408,6 +1409,12 @@ class ReposConfig(ConfigFile): return super().import_config(file, send_events) + + decision = self.import_decision() + if decision == SAVE: + self.save_database_instead() + return + repos_data = self._config.get('repos', []) # An empty repos list never deletes DB records. Deleting all repos is done through the API. @@ -1428,12 +1435,13 @@ class ReposConfig(ConfigFile): continue imported_ids.add(repo.id) - for repo in session.query(Repository).all(): + for repo in session.query(Repository).all() if decision == IMPORT else []: if repo.id not in imported_ids: logger.info(f'Deleting repo {repo.name!r} (no longer in config)') _delete_repository(session, repo) logger.info(f'Imported {len(imported_ids)} repos from {file_str}') + self.record_import_version() self.successful_import = True except Exception as e: self.successful_import = False @@ -1486,6 +1494,7 @@ class ReposConfig(ConfigFile): def dump_config(self, file: pathlib.Path = None, send_events=False, overwrite=False): """Dump all Repos to the config file.""" with get_db_session() as session: + version = self.db_version_for_dump(session) repos = session.query(Repository).options(*Repository.json_options()) \ .join(Collection, Collection.id == Repository.collection_id) \ .order_by(func.lower(Collection.name)).all() @@ -1508,6 +1517,7 @@ class ReposConfig(ConfigFile): data.append(entry) self._config['repos'] = data + self._config['version'] = version logger.info(f'Dumping {len(data)} repos to config') self.save(file, send_events, overwrite) @@ -1546,11 +1556,15 @@ class ReposConfig(ConfigFile): if i.get('url') and repo_url_key(i['url']) not in current_keys) self.write_config_data(dict(repos=merged, version=current_data.get('version', 0) + 1), config_file) - self.import_config(send_events=send_events) + # Importing a backup is the user's choice: apply it whatever the versions. + self.import_forced(send_events=send_events) save_downloads_config.activate_switch() repos_config = ReposConfig() +# Only the columns repos.yaml saves (not a fetch's commit, size, ...); its Collection saves the rest. +register_ownership(ReposConfig.file_name, Repository, + columns={'url', 'mode', 'branch', 'submodules', 'clone_token', 'collection_id'}) def get_repos_config() -> ReposConfig: diff --git a/modules/repos/test/test_config.py b/modules/repos/test/test_config.py index b6b98974..4d1c8863 100644 --- a/modules/repos/test/test_config.py +++ b/modules/repos/test/test_config.py @@ -3,6 +3,7 @@ import pytest from modules.repos import lib from modules.repos.models import Repository from wrolpi.collections.models import Collection +from wrolpi.conftest import config_versions_lost from wrolpi.downloader import Download, DownloadFrequency @@ -31,9 +32,12 @@ async def test_dump_and_import(test_session, test_directory, async_client, test_ for repo in test_session.query(Repository).all(): lib._delete_repository(test_session, repo) test_session.commit() + config_versions_lost(test_session) assert test_session.query(Collection).count() == 0 config.import_config() + # Imported without deletes; the database is saved, which records its version. + await await_switches() test_session.expire_all() repos = {i.name: i for i in test_session.query(Repository)} assert repos['kiwix-tools'].directory == test_directory / 'repos/software/kiwix-tools' diff --git a/modules/repos/test/test_never.py b/modules/repos/test/test_never.py index 75f219f0..e3922cb2 100644 --- a/modules/repos/test/test_never.py +++ b/modules/repos/test/test_never.py @@ -9,6 +9,7 @@ import yaml from modules.repos import lib from modules.repos.models import Repository +from wrolpi.conftest import config_versions_lost from wrolpi.downloader import Download, DownloadFrequency, DownloadStatus @@ -146,6 +147,7 @@ async def test_never_config(test_session, test_directory, async_client, test_wro for download in test_session.query(Download).all(): test_session.delete(download) test_session.commit() + config_versions_lost(test_session) subprocess.run(['rm', '-rf', str(test_directory / 'repos/missing')], check=True) config.import_config() diff --git a/modules/repos/test/test_ownership.py b/modules/repos/test/test_ownership.py index 0ee5d871..e573d795 100644 --- a/modules/repos/test/test_ownership.py +++ b/modules/repos/test/test_ownership.py @@ -5,6 +5,7 @@ import pytest from modules.repos import lib from modules.repos.conftest import TEST_GIT_ENV +from wrolpi.conftest import config_versions_lost from wrolpi.downloader import Download, DownloadStatus @@ -395,6 +396,7 @@ async def test_clone_token_survives_the_config(test_session, async_client, git_r # Lose the database, restore it from the config. lib._delete_repository(test_session, repo) test_session.commit() + config_versions_lost(test_session) config.import_config() test_session.expire_all() restored = test_session.query(lib.Repository).one() diff --git a/modules/videos/lib.py b/modules/videos/lib.py index 952f6458..2a64b7aa 100644 --- a/modules/videos/lib.py +++ b/modules/videos/lib.py @@ -18,6 +18,7 @@ from wrolpi import dates, flags 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.config_version import IMPORT, SAVE, register_ownership from wrolpi.dates import Seconds, from_timestamp from wrolpi.db import get_db_curs, get_db_session, session_curs from wrolpi.downloader import Download, download_manager @@ -293,7 +294,13 @@ class ChannelsConfig(ConfigFile): self.successful_import = True return - super().import_config() + super().import_config(file) + + decision = self.import_decision() + if decision == SAVE: + self.save_database_instead() + return + try: channels = self.channels # Empty channels list = never delete DB records @@ -395,7 +402,7 @@ class ChannelsConfig(ConfigFile): with get_db_session(commit=True) as session: # Delete any Channels that were deleted from the config, along with their Collections. orphaned_collection_ids = [] - for channel in session.query(Channel): + for channel in session.query(Channel) if decision == IMPORT else []: if channel.id not in updated_channel_ids: logger.warning(f'Deleting {channel} because it is not in the config.') orphaned_collection_ids.append(channel.collection_id) @@ -435,6 +442,7 @@ class ChannelsConfig(ConfigFile): with get_db_session() as session: link_channel_and_downloads(session) + self.record_import_version() self.successful_import = True except Exception as e: self.successful_import = False @@ -450,6 +458,7 @@ class ChannelsConfig(ConfigFile): channel_import_logger.info('Dumping channels to config') with get_db_session() as session: + version = self.db_version_for_dump(session) # Get all channels, ordered by collection name channels = session.query(Channel).join(Collection).order_by(Collection.name).all() @@ -457,6 +466,7 @@ class ChannelsConfig(ConfigFile): channels_data = [channel.config_view() for channel in channels] self._config['channels'] = channels_data + self._config['version'] = version channel_import_logger.info(f'Dumping {len(channels_data)} channels to config') self.save(file, send_events, overwrite) @@ -515,10 +525,15 @@ class ChannelsConfig(ConfigFile): config_file.parent.mkdir(parents=True, exist_ok=True) self.write_config_data(merged_data, config_file) - self.import_config(send_events=send_events) + # Importing a backup is the user's choice: apply it whatever the versions. + self.import_forced(send_events=send_events) CHANNELS_CONFIG: ChannelsConfig = ChannelsConfig() +# Only the columns channels.yaml saves (not a refresh's counts, info_json, ...); its Collection saves the rest. +register_ownership(ChannelsConfig.file_name, Channel, + columns={'url', 'source_id', 'generate_posters', 'calculate_duration', 'download_missing_data', + 'collection_id'}) TEST_CHANNELS_CONFIG: ChannelsConfig = None diff --git a/modules/videos/test/test_channels_config.py b/modules/videos/test/test_channels_config.py index c1bf085d..fc1a401c 100644 --- a/modules/videos/test/test_channels_config.py +++ b/modules/videos/test/test_channels_config.py @@ -14,6 +14,7 @@ import yaml from modules.videos.lib import get_channels_config from modules.videos.models import Channel +from wrolpi.conftest import config_was_imported, config_versions_lost @pytest.mark.asyncio @@ -187,6 +188,7 @@ class TestChannelsConfigEdgeCases: # Reinitialize and import config.initialize() + config_was_imported(test_session, 'channels.yaml') config.import_config() assert config.successful_import is True @@ -317,6 +319,7 @@ class TestChannelsConfigEdgeCases: # Clear the file_format to simulate factory reset channel.collection.file_format = None test_session.commit() + config_versions_lost(test_session) # Re-import config.initialize() diff --git a/modules/videos/test/test_common.py b/modules/videos/test/test_common.py index 0e76f188..67537497 100644 --- a/modules/videos/test/test_common.py +++ b/modules/videos/test/test_common.py @@ -9,7 +9,7 @@ from PIL import Image from modules.videos.models import Channel, Video from wrolpi.collections import Collection -from wrolpi.conftest import probe_write_lock_is_held, production_like_sessions +from wrolpi.conftest import config_was_imported, probe_write_lock_is_held, production_like_sessions from wrolpi.common import get_absolute_media_path, get_wrolpi_config, get_relative_to_media_directory from wrolpi.downloader import Download, DownloadFrequency from wrolpi.vars import PROJECT_DIR @@ -254,7 +254,8 @@ async def test_import_channel_downloads(await_switches, test_session, channel_fa [i.delete(add_to_skip_list=False) for i in test_session.query(Download).all()] # Write config with Channel Downloads now that the delete() has removed them from the config. get_channels_config().channels = channels_backup - import_channels_config() + # Deleting the Downloads made the database newer than the file, so restoring from the file is the user's choice. + get_channels_config().import_forced() channel1, channel2 = test_session.query(Channel).order_by(Channel.url).all() assert test_session.query(Download).count() == 3 assert channel1.url == 'https://example.com/channel1' @@ -353,6 +354,7 @@ async def test_import_channel_deletes_orphaned_collections(await_switches, test_ await await_switches() # Import config - this should delete channel2 AND its Collection + config_was_imported(test_session, 'channels.yaml') import_channels_config() # Verify only one Channel remains diff --git a/modules/zim/models.py b/modules/zim/models.py index 441d5829..f07292e7 100644 --- a/modules/zim/models.py +++ b/modules/zim/models.py @@ -13,6 +13,7 @@ from sqlalchemy.orm.exc import NoResultFound # noqa from modules.zim.errors import UnknownZimEntry, UnknownZimTagEntry, UnknownZim from wrolpi import dates, tags from wrolpi.common import loaded, Base, logger, get_relative_to_media_directory, ModelHelper +from wrolpi.config_version import register_ownership from wrolpi.dates import TZDateTime from wrolpi.db import session_curs, serializer from wrolpi.downloader import Download, download_manager @@ -380,6 +381,9 @@ class TagZimEntry(Base): .one_or_none() +register_ownership('tags.yaml', TagZimEntry) + + class ZimSubscription(Base): __tablename__ = 'zim_subscription' diff --git a/wrolpi/collections/config.py b/wrolpi/collections/config.py index 93744147..91beaa86 100644 --- a/wrolpi/collections/config.py +++ b/wrolpi/collections/config.py @@ -4,6 +4,7 @@ from dataclasses import dataclass, field from typing import List from wrolpi.common import ConfigFile, get_media_directory, get_relative_to_media_directory, logger +from wrolpi.config_version import IMPORT, SAVE from wrolpi.db import get_db_session from wrolpi.events import Events from wrolpi.switches import ActivateSwitchMethod, register_switch_handler @@ -13,6 +14,11 @@ logger = logger.getChild(__name__) __all__ = ['PlaylistsConfig', 'playlists_config', 'save_playlists_config', 'get_playlists_config'] +def _item_key(item) -> tuple: + """What a playlist item points to; a playlist has at most one item per key (see CollectionItem's constraints).""" + return item.item_kind, item.file_group_id, item.zim_id, item.zim_entry, item.url + + @dataclass class PlaylistsConfigValidator: """Validator for the playlists config file.""" @@ -69,6 +75,12 @@ class PlaylistsConfig(ConfigFile): return super().import_config(file, send_events) + + decision = self.import_decision() + if decision == SAVE: + self.save_database_instead() + return + playlists_data = self._config.get('playlists', []) # An empty playlists list never deletes DB records (mirrors ChannelsConfig). Deleting all @@ -90,7 +102,8 @@ class PlaylistsConfig(ConfigFile): collection = session.query(Collection).filter_by( name=name, kind='playlist').one_or_none() - if not collection: + new = not collection + if new: collection = Collection(name=name, kind='playlist') session.add(collection) collection.description = data.get('description') @@ -103,12 +116,7 @@ class PlaylistsConfig(ConfigFile): collection.tag = Tag.get_by_name(session, tag_name) if tag_name else None session.flush([collection]) - # Rebuild items to match the config's order exactly. - for item in list(collection.items): - session.delete(item) - session.flush() - - position = 1 + file_items = [] for item_data in (data.get('items') or []): item = None if 'file' in item_data: @@ -136,20 +144,43 @@ class PlaylistsConfig(ConfigFile): url=item_data['url']) if item is not None: item.title = item_data.get('title') - item.position = position - position += 1 - session.add(item) + file_items.append(item) + + if decision == IMPORT or new: + # Rebuild items to match the config's order exactly. + for item in list(collection.items): + session.delete(item) + session.flush() + items = file_items + else: + # Without deletes, keep the database's items in their order; add the file's other items + # after the item that precedes them in the file. + items = sorted(collection.items, key=lambda i: i.position) + existing = {_item_key(i): i for i in items} + previous = None + for item in file_items: + if found := existing.get(_item_key(item)): + previous = found + continue + items.insert(items.index(previous) + 1 if previous else 0, item) + previous = item + + for position, item in enumerate(items, start=1): + item.position = position + session.add(item) session.flush() imported_names.add(name) # Delete playlists that are no longer present in the config. - for collection in session.query(Collection).filter_by(kind='playlist').all(): + for collection in session.query(Collection).filter_by(kind='playlist').all() \ + if decision == IMPORT else []: if collection.name not in imported_names: logger.info(f'Deleting playlist {collection.name!r} (no longer in config)') session.delete(collection) logger.info(f'Imported {len(imported_names)} playlists from {file_str}') + self.record_import_version() self.successful_import = True except Exception as e: self.successful_import = False @@ -217,7 +248,8 @@ class PlaylistsConfig(ConfigFile): config_file.parent.mkdir(parents=True, exist_ok=True) self.write_config_data(merged_data, config_file) - self.import_config(send_events=send_events) + # Importing a backup is the user's choice: apply it whatever the versions. + self.import_forced(send_events=send_events) # Materialize the restored playlists on disk. from .sync import sync_playlists_directory sync_playlists_directory.activate_switch() @@ -226,20 +258,9 @@ class PlaylistsConfig(ConfigFile): """Dump all playlists (with their ordered items) to the config file.""" from .models import Collection - # playlists.yaml is derived from the database (the source of truth here), so the dump should - # always win. Sync our in-memory version up to the on-disk version first, otherwise a Sanic - # worker (or freshly-restarted process) that hasn't re-imported would refuse to overwrite a - # newer on-disk config. - target = file or self.get_file() - try: - if target.exists(): - disk_version = (self.read_config_file(target) or {}).get('version') or 0 - if disk_version > (self._config.get('version') or 0): - self._config['version'] = disk_version - except Exception as e: - logger.warning(f'Could not read existing playlists config version: {e}') - with get_db_session() as session: + # The database's version (see wrolpi.config_version); it is never older than a file it saved. + version = self.db_version_for_dump(session) playlists = session.query(Collection).filter_by(kind='playlist') \ .order_by(Collection.name).all() data = [] @@ -255,6 +276,7 @@ class PlaylistsConfig(ConfigFile): entry['items'] = [c for c in (i.to_config() for i in collection.items) if c] data.append(entry) self._config['playlists'] = data + self._config['version'] = version logger.info(f'Dumping {len(data)} playlists to config') self.save(file, send_events, overwrite) diff --git a/wrolpi/collections/models.py b/wrolpi/collections/models.py index cb6041a3..047cb35c 100644 --- a/wrolpi/collections/models.py +++ b/wrolpi/collections/models.py @@ -12,6 +12,7 @@ from sqlalchemy.orm import Mapped, relationship, Session, joinedload, selectinlo from wrolpi import flags from wrolpi.common import loaded, Base, ModelHelper, logger, get_media_directory, get_relative_to_media_directory, \ unique_by_predicate, TRACE_LEVEL +from wrolpi.config_version import collection_kinds, count_config_change, register_ownership from wrolpi.db import serializer from wrolpi.downloader import Download, save_downloads_config from wrolpi.errors import ValidationError @@ -734,6 +735,9 @@ class Collection(ModelHelper, Base): ).delete(synchronize_session=False) logger.debug(f'Removed {deleted} FileGroups from Collection {self.id}') + if deleted and self.kind == 'playlist': + # A bulk delete is not flushed: count it. + count_config_change(session, 'playlists.yaml') # Resequence positions to close gaps (optional, but keeps positions clean) items = session.query(CollectionItem).filter_by( @@ -1310,3 +1314,13 @@ class CollectionItem(ModelHelper, Base): if self.title: data['title'] = self.title return data + + +# Each kind of Collection is saved by its own config, which saves these columns (not item_count, total_size, ...). +COLLECTION_CONFIG_COLUMNS = {'name', 'kind', 'description', 'directory', 'tag_id', 'file_format'} +for _kind, _config in (('channel', 'channels.yaml'), ('domain', 'domains.yaml'), ('playlist', 'playlists.yaml'), + ('repo', 'repos.yaml')): + register_ownership(_config, Collection, columns=COLLECTION_CONFIG_COLUMNS, + row=lambda collection, kind=_kind: collection.kind == kind) +# Only a playlist's items are saved (author and subject Collections have items too). +register_ownership('playlists.yaml', CollectionItem, row=lambda item: 'playlist' in collection_kinds(item)) diff --git a/wrolpi/collections/test/test_domains_config.py b/wrolpi/collections/test/test_domains_config.py index ee1abbc3..49061e39 100644 --- a/wrolpi/collections/test/test_domains_config.py +++ b/wrolpi/collections/test/test_domains_config.py @@ -9,6 +9,7 @@ from sqlalchemy.orm import Session from modules.archive.lib import DomainsConfig, domains_config from wrolpi.collections import Collection +from wrolpi.conftest import config_was_imported class TestDomainsConfig: @@ -144,6 +145,7 @@ class TestDomainsConfig: config_file.write_text(yaml.dump(config_data)) config = DomainsConfig() + config_was_imported(test_session, 'domains.yaml') config.import_config(file=config_file) # test.org should be deleted @@ -317,6 +319,7 @@ async def test_import_keeps_domains_with_archives(test_session: Session, test_di test_session.commit() assert test_session.query(Archive).count() == 2 + config_was_imported(test_session, 'domains.yaml') domains_config.import_config() test_session.expire_all() assert {i.name for i in test_session.query(Collection)} == {'kept.com', 'empty.com', 'new.com'} diff --git a/wrolpi/collections/test/test_playlists_config.py b/wrolpi/collections/test/test_playlists_config.py index 5af1e74c..474c0507 100644 --- a/wrolpi/collections/test/test_playlists_config.py +++ b/wrolpi/collections/test/test_playlists_config.py @@ -4,9 +4,11 @@ from copy import deepcopy from http import HTTPStatus import pytest +import yaml from wrolpi.collections.config import get_playlists_config from wrolpi.collections.models import Collection +from wrolpi.conftest import config_was_imported @pytest.fixture @@ -233,6 +235,7 @@ async def test_import_still_deletes_playlists_removed_from_config( ' - url: /map?lat=1&lon=2&z=3\n' ) + config_was_imported(test_session, 'playlists.yaml') playlists_config.import_config() test_session.expire_all() @@ -368,3 +371,24 @@ async def test_playlist_changes_wait_for_config_import(test_session, test_direct playlists_config.import_config() request, response = await async_client.post('/api/collections', content=json.dumps(dict(name='New'))) assert response.status_code == HTTPStatus.CREATED, response.json + + +@pytest.mark.asyncio +async def test_import_without_deletes_keeps_database_items(test_session, test_directory, playlists_config): + """Without a database version (e.g. the first startup after an upgrade) a playlist's items in the database are + kept in their order, and the file's other items are added after the item that precedes them in the file.""" + collection = Collection(name='Mix', kind='playlist') + test_session.add(collection) + test_session.flush([collection]) + collection.add_url(test_session, '/a', title='A') + collection.add_url(test_session, '/c', title='C') # Added after the file was saved. + test_session.commit() + playlists_config.get_file().write_text(yaml.dump(dict(version=1, playlists=[ + dict(name='Mix', items=[dict(url='/b', title='B'), dict(url='/a', title='A')]), + ]))) + + playlists_config.import_config() + + test_session.expire_all() + collection = test_session.query(Collection).filter_by(name='Mix', kind='playlist').one() + assert [i.url for i in collection.items] == ['/b', '/a', '/c'] diff --git a/wrolpi/common.py b/wrolpi/common.py index c18a7736..c04d60ff 100644 --- a/wrolpi/common.py +++ b/wrolpi/common.py @@ -723,7 +723,9 @@ class ConfigFile: logger_.debug(f'Copied backup config: {rel_path} -> {backup_file}') # Write the config in-memory to the file. Track version changes to avoid overriding newer config. - self._config['version'] = (self._config['version'] or 0) + 1 + # A database config's version is the database's, which `dump_config` set (see wrolpi.config_version). + if not self.is_versioned_by_database(): + self._config['version'] = (self._config['version'] or 0) + 1 config = deepcopy(self._config) self.write_config_data(config, file) @@ -739,6 +741,9 @@ class ConfigFile: Events.send_config_save_failed(message) raise e + if self.is_versioned_by_database() and file == self.get_file(): + self.record_saved_version(config['version']) + def write_config_data(self, config: dict, config_file: pathlib.Path): write_config_data(config, config_file, width=self.width) @@ -782,6 +787,86 @@ class ConfigFile: # Copy the data as-is by default. Other classes will overwrite this and dump the database before saving. self.save(file, send_events, overwrite) + # A database config's file and the database each carry a version; see wrolpi.config_version. + + # The Import button sets this: the user chose to apply the file, whatever its version. + force_import: bool = False + + def import_forced(self, send_events=False): + """Import this config's file whatever its version: the user chose it (the Import button, a backup).""" + self.force_import = True + try: + self.import_config(send_events=send_events) + finally: + # An import that returned before `import_decision` must not force a later one. + self.force_import = False + + def is_versioned_by_database(self) -> bool: + from wrolpi.config_version import versioned_config_names + return self.file_name in versioned_config_names() + + # What the last `import_decision` allowed; `record_import_version` records the version that follows from it. + last_import_decision: str = None + + def import_decision(self) -> str: + """After reading the file (`ConfigFile.import_config`), decide what importing it may do: IMPORT, + IMPORT_WITHOUT_DELETES, or SAVE (the database is newer; call `save_database_instead` and import nothing). + + On IMPORT the database's version is removed until `record_import_version`: the import's own writes must not + count, or an import that fails partway would leave the database "newer" than the file it did not finish + applying, and the next startup would save that partial database over the file.""" + from wrolpi.config_version import IMPORT, decide, delete_config_version, get_config_version + from wrolpi.db import get_db_session + with get_db_session(commit=True) as session: + db_version = get_config_version(session, self.file_name) + if self.force_import: + self.force_import = False + decision = IMPORT + else: + decision = decide(self.version or 0, db_version) + if decision == IMPORT: + delete_config_version(session, self.file_name) + logger_.debug(f'{self.file_name}: file version {self.version}, database version {db_version}: {decision}') + self.last_import_decision = decision + return decision + + 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') + self.successful_import = True + self.dump_config() + + def record_import_version(self): + """The database now matches this config's file: record the file's version. Call last, after every write + the import makes. + + Not after IMPORT_WITHOUT_DELETES: the database may keep rows the file does not list (changes that were never + saved), so recording the file's version would let the next import delete them. Save them into the file + instead; that save records the version (`record_saved_version`). A save after a partial import is refused, + so the version stays unknown until an import applies the whole file.""" + from wrolpi.config_version import IMPORT, set_config_version + from wrolpi.db import get_db_session + if self.last_import_decision != IMPORT: + self.background_dump.activate_switch() + return + with get_db_session(commit=True) as session: + set_config_version(session, self.file_name, self.version or 0) + + def db_version_for_dump(self, session) -> int: + """The version `dump_config` writes: the database's, read in the same transaction as the rows it saves. + Before the database has one, the file's next version (`save` records it).""" + from wrolpi.config_version import get_config_version + db_version = get_config_version(session, self.file_name) + return db_version if db_version is not None else (self.version or 0) + 1 + + def record_saved_version(self, version: int): + """The file was just written from the database. If the database's version was unknown (see + `record_import_version`), the two are now in step: record it, so changes from here on count against it.""" + from wrolpi.config_version import create_config_version + from wrolpi.db import get_db_session + with get_db_session(commit=True) as session: + create_config_version(session, self.file_name, version) + def update(self, config: dict, overwrite: bool = False): """Update any values of this config. Save the config to its file.""" from wrolpi.api_utils import api_app diff --git a/wrolpi/config_api.py b/wrolpi/config_api.py index cfa233db..6f23424c 100644 --- a/wrolpi/config_api.py +++ b/wrolpi/config_api.py @@ -49,7 +49,8 @@ def post_config_update(request: Request, body: config_schema.ConfigUpdateRequest def post_config_import(_: Request, body: config_schema.ConfigsRequest): config = get_config_by_file_name(body.file_name) try: - config.import_config(send_events=True) + # The user chose to apply the file, whatever its version. + config.import_forced(send_events=True) except Exception as e: logger.error(f'Failed to import config: {body.file_name}', exc_info=e) raise InvalidConfig(f'Failed to import config {body.file_name}') diff --git a/wrolpi/config_version.py b/wrolpi/config_version.py new file mode 100644 index 00000000..c56e371e --- /dev/null +++ b/wrolpi/config_version.py @@ -0,0 +1,185 @@ +"""Which saved version of each database config the database is in step with. + +A database config (tags.yaml, channels.yaml, ...) saves rows the database owns. Its file carries an integer +`version`; the database keeps one too, in `config_version`. Every change to rows a config owns adds one to that +config's database version, in the same transaction (`_count_config_changes`). `dump_config` writes the database's +version into the file it saves, and an import records the file's version once it has applied it. + +So an import can tell which side is newer by comparing two integers (see `ConfigFile.import_decision`): + +* file == database: in step. The file is current; import it, deleting what it no longer lists. +* file > database: the file is newer (a restored or wiped database, a copied config); import it the same way. +* file < database: the database changed and was not saved yet (e.g. WROLPi restarted first). Save the database + instead of importing the older file, which would undo those changes. +* no database row: not known yet (a config never imported by this code, or an import that did not finish). + Import without deleting, and leave the version unknown: the database may have rows the file does not list. Then + save them into the file; that save records its version (`ConfigFile.record_saved_version`). + +Integers only: a Raspberry Pi has no clock that survives a reboot, so timestamps cannot order these. +""" +import json +from collections import defaultdict +from dataclasses import dataclass +from typing import Callable, Dict, FrozenSet, List, Optional, Set, Type + +import sqlite3 + +from sqlalchemy import Column, Integer, String, event, inspect, text +from sqlalchemy.orm import Session + +from wrolpi.common import Base + + +class ConfigVersion(Base): + __tablename__ = 'config_version' + name = Column(String, primary_key=True) # The config's file_name, e.g. "tags.yaml". + version = Column(Integer, nullable=False) + + def __repr__(self): + return f'' + + +@dataclass(frozen=True) +class Ownership: + config: str + model: Type + # Only changes to these columns count; None counts every column. + columns: Optional[FrozenSet[str]] = None + # Only rows this accepts belong to the config (e.g. Collections of one kind); None accepts every row. + row: Optional[Callable] = None + + +OWNERSHIPS: Dict[Type, List[Ownership]] = defaultdict(list) + + +def register_ownership(config: str, model: Type, columns: Set[str] = None, row: Callable = None): + """Declare that `config` (a file_name) saves `model`'s rows, so changing them makes the database newer than + the config file until it is saved again.""" + OWNERSHIPS[model].append(Ownership(config, model, frozenset(columns) if columns else None, row)) + + +def versioned_config_names() -> Set[str]: + return {o.config for ownerships in OWNERSHIPS.values() for o in ownerships} + + +def _owners(obj, changed: bool) -> Set[str]: + """The configs that save `obj`. `changed`: only those whose columns of an updated row changed; otherwise (an + inserted or deleted row) only those whose columns have a value.""" + names = set() + for ownership in OWNERSHIPS.get(type(obj), ()): + # Columns first: `row` may query (e.g. `collection_kind`). + if changed: + state = inspect(obj) + columns = ownership.columns or [i.key for i in state.mapper.column_attrs] + if not any(state.attrs[i].history.has_changes() for i in columns): + continue + elif ownership.columns and all(getattr(obj, i) is None for i in ownership.columns): + # e.g. a FileGroup that was never viewed is not in recently_viewed.yaml. + continue + if ownership.row and not ownership.row(obj): + continue + names.add(ownership.config) + return names + + +_COLLECTION_KINDS = 'config_version_collection_kinds' + + +@event.listens_for(Session, 'after_flush') +def _count_config_changes(session: Session, flush_context): + """Add one to the database version of each config whose rows this flush changed, in the same transaction. + + Only existing rows are counted: a config without a row is not known yet, and its next import creates the row. + Raw SQL writes to owned tables are not seen here; they must call `count_config_change`.""" + if not OWNERSHIPS: + return + names = set() + session.info[_COLLECTION_KINDS] = dict() + try: + for obj in session.new: + names |= _owners(obj, changed=False) + for obj in session.deleted: + names |= _owners(obj, changed=False) + for obj in session.dirty: + names |= _owners(obj, changed=True) + finally: + del session.info[_COLLECTION_KINDS] + if names: + count_config_change(session, *names) + + +def count_config_change(session: Session, *names: str): + """Add one to these configs' database versions (only those which have one).""" + params = {f'n{i}': name for i, name in enumerate(names)} + placeholders = ', '.join(f':{i}' for i in params) + session.connection().execute( + text(f'UPDATE config_version SET version = version + 1 WHERE name IN ({placeholders})'), params) + + +def count_config_change_curs(curs, *names: str): + """`count_config_change` for a raw `sqlite3` cursor, in its transaction.""" + placeholders = ', '.join('?' for _ in names) + try: + curs.execute(f'UPDATE config_version SET version = version + 1 WHERE name IN ({placeholders})', names) + except sqlite3.OperationalError as e: + # A migration older than config_version runs this code too. + if 'no such table' not in str(e): + raise + + +def collection_kinds(obj) -> Set[str]: + """The kinds of the Collections `obj` (e.g. a Download) belongs to, and belonged to before this flush (moving + it out of a Collection changes that Collection's config too), read on its flush's connection.""" + ids = {i for i in inspect(obj).attrs.collection_id.history.deleted if i} + if obj.collection_id: + ids.add(obj.collection_id) + if not ids: + return set() + session = Session.object_session(obj) + # A Collection's kind never changes: look each up once per flush (a playlist import writes thousands of items). + kinds = session.info.get(_COLLECTION_KINDS, dict()) + if missing := ids - kinds.keys(): + rows = session.connection().execute( + text('SELECT id, kind FROM collection WHERE id IN (SELECT value FROM json_each(:ids))'), + {'ids': json.dumps(sorted(missing))}) + kinds.update(dict.fromkeys(missing)) + kinds.update(rows.all()) + return {kinds[i] for i in ids} - {None} + + +def get_config_version(session: Session, name: str) -> Optional[int]: + row = session.connection().execute( + text('SELECT version FROM config_version WHERE name = :name'), {'name': name}).first() + return row[0] if row else None + + +def delete_config_version(session: Session, name: str): + session.connection().execute(text('DELETE FROM config_version WHERE name = :name'), {'name': name}) + + +def create_config_version(session: Session, name: str, version: int): + """Record `version` unless the database already has one.""" + session.connection().execute( + text('INSERT INTO config_version (name, version) VALUES (:name, :version) ON CONFLICT (name) DO NOTHING'), + {'name': name, 'version': version}) + + +def set_config_version(session: Session, name: str, version: int): + session.connection().execute( + text('INSERT INTO config_version (name, version) VALUES (:name, :version) ' + 'ON CONFLICT (name) DO UPDATE SET version = excluded.version'), + {'name': name, 'version': version}) + + +IMPORT = 'import' +IMPORT_WITHOUT_DELETES = 'import_without_deletes' +SAVE = 'save' + + +def decide(file_version: int, db_version: Optional[int]) -> str: + """The table in this module's docstring.""" + if db_version is None: + return IMPORT_WITHOUT_DELETES + if file_version < db_version: + return SAVE + return IMPORT diff --git a/wrolpi/conftest.py b/wrolpi/conftest.py index 364184d4..906b7ed0 100644 --- a/wrolpi/conftest.py +++ b/wrolpi/conftest.py @@ -131,6 +131,24 @@ def no_unexpected_queries(request): '\n'.join(f'{reason}: {" ".join(statement.split())[:300]}' for reason, statement in queries) +def config_was_imported(session, *file_names: str): + """Record that these database configs were imported before, as on any WROLPi after its first startup with + wrolpi.config_version, so their next import deletes what the file no longer lists. (Version 0 is no newer than + any file.) Call it after the test's own database changes: a change counts as newer than the file.""" + from wrolpi.config_version import set_config_version + for file_name in file_names: + set_config_version(session, file_name, 0) + session.commit() + + +def config_versions_lost(session): + """A test that deletes rows to simulate a lost database must lose the database's config versions too (a real + lost database has none); otherwise deleting the rows counts as a database change newer than the file.""" + from wrolpi.config_version import ConfigVersion + session.query(ConfigVersion).delete() + session.commit() + + @pytest.fixture(autouse=True) def no_deferred_request_writes(request): """Fail any test during which an API request wrote outside `request.ctx.db.write()`.""" diff --git a/wrolpi/downloader.py b/wrolpi/downloader.py index ff6a5cf8..bd51c21f 100644 --- a/wrolpi/downloader.py +++ b/wrolpi/downloader.py @@ -24,9 +24,9 @@ from zoneinfo import ZoneInfo import feedparser import pytz from feedparser import FeedParserDict -from sqlalchemy import Column, Integer, String, Text, ForeignKey, Index, JSON +from sqlalchemy import Column, Integer, String, Text, ForeignKey, Index, JSON, text from sqlalchemy.exc import OperationalError -from sqlalchemy.orm import Mapped, Session, relationship +from sqlalchemy.orm import Mapped, Session, column_property, relationship from wrolpi import flags from wrolpi.api_utils import api_app, perpetual_signal @@ -34,6 +34,7 @@ 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.dates import TZDateTime, now, Seconds from wrolpi.db import get_db_session, session_curs, serializer from wrolpi.errors import InvalidDownload, UnrecoverableDownloadError, BotBlockedDownloadError, UnknownDownload, \ @@ -268,7 +269,9 @@ class Download(ModelHelper, Base): # noqa tag_names = Column(JSON) # a list of Tag names. # A Download may be associated with a Collection (downloads for channels, domains, etc.). - collection_id = Column(Integer, ForeignKey('collection.id', ondelete='SET NULL')) + # The previous value is loaded on change: leaving a Collection changes that Collection's config too. + collection_id = column_property(Column(Integer, ForeignKey('collection.id', ondelete='SET NULL')), + active_history=True) collection = relationship('Collection', primaryjoin='Download.collection_id==Collection.id', back_populates='downloads') @@ -1317,9 +1320,12 @@ class DownloadManager: @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) - session.query(Download) \ + retried = session.query(Download) \ .filter(Download.status.in_((DownloadStatus.pending, DownloadStatus.deferred, DownloadStatus.failed))) \ .update(values, synchronize_session=False) + if retried: + # A bulk update is not flushed: count it. `status` is saved only by download_manager.yaml. + count_config_change(session, DownloadManagerConfig.file_name) def retry_downloads_on_startup(self): """Startup is not a request: retry incomplete Downloads in a short write transaction of its own.""" @@ -1751,6 +1757,8 @@ class DownloadManager: delete = delete.where(Download.status == status) values = [i for i, in query.all()] session.execute(delete) + # A Core delete is not flushed, so count it for download_manager.yaml (wrolpi.config_version). + count_config_change(session, DownloadManagerConfig.file_name) return values def delete_completed(self, session: Session) -> List[int]: @@ -1781,6 +1789,8 @@ class DownloadManager: if not download_ids: return [] deleted_ids = [i for i, in session.query(Download.id).filter(Download.id.in_(download_ids)).all()] + # A Core delete is not flushed: count it. + count_download_changes(session, deleted_ids) session.execute(Download.__table__.delete().where(Download.id.in_(deleted_ids))) session.commit() save_downloads_config.activate_switch() @@ -2074,6 +2084,7 @@ class DownloadManagerConfig(ConfigFile): def dump_config(self, file: pathlib.Path = None, send_events=False, overwrite=False): try: with get_db_session() as session: + version = self.db_version_for_dump(session) downloads: Iterable[Download] = session.query(Download).order_by(Download.url) new_downloads = [] for download in downloads: @@ -2094,7 +2105,8 @@ class DownloadManagerConfig(ConfigFile): url=download.url, ) new_downloads.append(new_download) - get_download_manager_config().update({'downloads': new_downloads}, overwrite=overwrite) + get_download_manager_config().update({'downloads': new_downloads, 'version': version}, + overwrite=overwrite) except Exception as e: message = f'Failed to save {self.get_relative_file()} config' logger.error(message, exc_info=e) @@ -2112,6 +2124,11 @@ class DownloadManagerConfig(ConfigFile): 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: @@ -2155,7 +2172,7 @@ class DownloadManagerConfig(ConfigFile): # 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 - else: + elif decision == IMPORT: logger.warning(f'Deleting Download {existing.url} because it is no longer in the config') session.delete(existing) @@ -2205,6 +2222,7 @@ class DownloadManagerConfig(ConfigFile): if need_commit: session.commit() + self.record_import_version() self.successful_import = True except Exception as e: self.successful_import = False @@ -2287,10 +2305,35 @@ class DownloadManagerConfig(ConfigFile): config_file.parent.mkdir(parents=True, exist_ok=True) self.write_config_data(merged_data, config_file) - self.import_config(send_events=send_events) + # Importing a backup is the user's choice: apply it whatever the versions. + self.import_forced(send_events=send_events) DOWNLOAD_MANAGER_CONFIG: DownloadManagerConfig = DownloadManagerConfig() +# Each config owns only the columns it saves: counting another change would make its next import save instead of +# import. (`attempts`, `error`, ... change on every attempt.) +register_ownership(DownloadManagerConfig.file_name, Download, + columns={'destination', 'downloader', 'frequency', 'last_successful_download', 'next_download', + 'settings', 'status', 'sub_downloader', 'tag_names', 'url'}) +# channels.yaml, domains.yaml and repos.yaml each save their own Collections' Downloads (the URL and how often it +# runs). Only the config of that Collection's kind owns one. +DOWNLOAD_COLLECTION_CONFIGS = (('channel', 'channels.yaml'), ('domain', 'domains.yaml'), ('repo', 'repos.yaml')) +for _kind, _config in DOWNLOAD_COLLECTION_CONFIGS: + register_ownership(_config, Download, columns={'url', 'frequency', 'collection_id'}, + row=lambda download, kind=_kind: kind in collection_kinds(download)) + + +def count_download_changes(session: Session, download_ids: List[int]): + """Count a write that bypasses the flush (Core or bulk) to these Downloads, before it runs: for + download_manager.yaml, and for the config of each Collection kind they belong to.""" + if not download_ids: + return + kinds = {i for i, in session.execute( + text('SELECT DISTINCT c.kind FROM download d JOIN collection c ON c.id = d.collection_id ' + 'WHERE d.id IN (SELECT value FROM json_each(:ids))'), + {'ids': json.dumps(list(download_ids))})} + count_config_change(session, DownloadManagerConfig.file_name, + *(config for kind, config in DOWNLOAD_COLLECTION_CONFIGS if kind in kinds)) TEST_DOWNLOAD_MANAGER_CONFIG: DownloadManagerConfig = None diff --git a/wrolpi/files/lib.py b/wrolpi/files/lib.py index 3fab2892..9240a38d 100644 --- a/wrolpi/files/lib.py +++ b/wrolpi/files/lib.py @@ -31,6 +31,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.config_version import count_config_change_curs 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, \ @@ -1067,6 +1068,8 @@ def merge_file_group_relations(curs, loser_id: int, winner_id: int): FROM collection_item WHERE file_group_id = ?)) ''', (winner_id, loser_id, winner_id)) + # Raw SQL: count these changes for the configs that save TagFiles and playlist items (wrolpi.config_version). + count_config_change_curs(curs, 'tags.yaml', 'playlists.yaml') def merge_into_surviving_duplicates(curs, file_group_ids: List[int], deleting_ids: Set[int]) -> int: @@ -2416,6 +2419,7 @@ def _bulk_insert_tag_files(tag_files: List[Tuple[int, int]]): ON CONFLICT (tag_id, file_group_id) DO NOTHING ''' curs.execute(stmt, params) + count_config_change_curs(curs, 'tags.yaml') logger.debug(f'Bulk inserted {len(chunk)} TagFiles') @@ -2441,6 +2445,7 @@ def _bulk_delete_tag_files(tag_files: List[Tuple[int, int]]): ''' curs.execute(stmt, params) logger.debug(f'Bulk deleted {curs.rowcount} TagFiles') + count_config_change_curs(curs, 'tags.yaml') async def _process_bulk_tag_job(job: dict): diff --git a/wrolpi/files/models.py b/wrolpi/files/models.py index 2d9955b5..6c3635c1 100644 --- a/wrolpi/files/models.py +++ b/wrolpi/files/models.py @@ -13,6 +13,7 @@ from sqlalchemy.orm import Mapped, deferred, relationship, Session, selectinload from wrolpi.common import loaded, Base, ModelHelper, logger, recursive_map, get_media_directory, \ get_relative_to_media_directory, unique_by_predicate, replace_file from wrolpi.dates import TZDateTime, now, from_timestamp, strptime_ms, strftime +from wrolpi.config_version import count_config_change, register_ownership from wrolpi.db import get_db_session, serializer from wrolpi.downloader import Download from wrolpi.errors import FileGroupIsTagged, UnknownFile @@ -251,6 +252,8 @@ class FileGroup(ModelHelper, Base): session.flush(self.tag_files) if deleted_tags := (existing_tag_ids - tag_ids): + # A bulk delete is not flushed: count it. + count_config_change(session, 'tags.yaml') session.query(TagFile).filter( TagFile.file_group_id == self.id, TagFile.tag_id.in_(deleted_tags), @@ -885,6 +888,10 @@ def update_effective_datetime(mapper, connection, target): target.effective_datetime = target.published_datetime or target.download_datetime +# recently_viewed.yaml saves only how far each viewed file was read or watched. +register_ownership('recently_viewed.yaml', FileGroup, columns={'viewed', 'progress', 'position'}) + + class Directory(ModelHelper, Base): """A representation of a file directory in the media directory.""" __tablename__ = 'directory' diff --git a/wrolpi/files/progress.py b/wrolpi/files/progress.py index c0cdf342..b831e7f9 100644 --- a/wrolpi/files/progress.py +++ b/wrolpi/files/progress.py @@ -24,6 +24,7 @@ import pytz from wrolpi import dates from wrolpi.common import ConfigFile, logger, get_media_directory, get_paths_in_media_directory +from wrolpi.config_version import SAVE from wrolpi.db import get_db_session from wrolpi.errors import ValidationError, NoPrimaryFile from wrolpi.events import Events @@ -151,6 +152,7 @@ class RecentlyViewedConfig(ConfigFile): from wrolpi.files.models import FileGroup media_directory = get_media_directory() with get_db_session() as session: + version = self.db_version_for_dump(session) file_groups = session.query(FileGroup) \ .filter(FileGroup.viewed.isnot(None)) \ .order_by(FileGroup.viewed.desc(), FileGroup.id) \ @@ -171,7 +173,7 @@ class RecentlyViewedConfig(ConfigFile): if file_group.position: entry['position'] = file_group.position files.append(entry) - self.update({'files': files}, overwrite=overwrite) + self.update({'files': files, 'version': version}, overwrite=overwrite) except Exception as e: message = f'Failed to save {self.get_relative_file()} config' logger.error(message, exc_info=e) @@ -182,6 +184,9 @@ class RecentlyViewedConfig(ConfigFile): """Apply the config to the database. Progress written after the config (e.g. while WROLPi was starting) is kept; a newer plain view keeps its `viewed`, but takes the config's position.""" super().import_config(file, send_events) + if self.import_decision() == SAVE: + self.save_database_instead() + return try: # wrolpi.files.models imports this module. from wrolpi.files.lib import glob_shared_stem @@ -238,6 +243,7 @@ class RecentlyViewedConfig(ConfigFile): applied += 1 logger.info(f'Imported {applied} of {len(entries)} recently viewed files') + self.record_import_version() self.successful_import = True except Exception as e: self.successful_import = False diff --git a/wrolpi/files/worker.py b/wrolpi/files/worker.py index 98e54b0c..941ee525 100644 --- a/wrolpi/files/worker.py +++ b/wrolpi/files/worker.py @@ -802,7 +802,7 @@ class FileWorker(FileMoveMixin, FileReorganizeMixin): operation_percent=0, ) - from wrolpi.downloader import Download, download_manager + from wrolpi.downloader import Download, count_download_changes, download_manager from wrolpi.files.models import FileGroup from wrolpi.tags import TagFile, save_tags_config, sync_tags_directory @@ -830,7 +830,9 @@ class FileWorker(FileMoveMixin, FileReorganizeMixin): for fg in file_groups: if fg.url: all_urls.append(fg.url) - if Download.get_by_url(session, fg.url): + if download := Download.get_by_url(session, fg.url): + # A bulk delete is not flushed: count it. + count_download_changes(session, [download.id]) session.query(Download).filter( Download.url == fg.url ).delete(synchronize_session=False) diff --git a/wrolpi/tags.py b/wrolpi/tags.py index 62469b09..f8139426 100644 --- a/wrolpi/tags.py +++ b/wrolpi/tags.py @@ -13,6 +13,7 @@ from sqlalchemy.orm import Mapped, relationship, Session from wrolpi import dates, flags from wrolpi.common import loaded, 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.config_version import IMPORT, SAVE, register_ownership from wrolpi.dates import TZDateTime from wrolpi.db import get_db_session, named_placeholders, session_curs, serializer from wrolpi.downloader import save_downloads_config @@ -337,6 +338,7 @@ class TagsConfig(ConfigFile): media_directory = get_media_directory() with get_db_session() as session: + version = self.db_version_for_dump(session) tags = dict() tag_rows = session.query(Tag) for tag in tag_rows: @@ -382,6 +384,7 @@ class TagsConfig(ConfigFile): 'tag_files': tag_files, 'tag_zims': tag_zims, 'tags': tags, + 'version': version, }) except Exception as e: message = f'Failed to save {self.get_relative_file()} config' @@ -403,6 +406,11 @@ class TagsConfig(ConfigFile): logger.info('Importing tags config') super().import_config(file) + decision = self.import_decision() + if decision == SAVE: + self.save_database_instead() + return + try: need_commit = False @@ -550,7 +558,7 @@ class TagsConfig(ConfigFile): self.import_skipped += 1 # Delete missing Tags last in case they are used above. - if config.tags: + if config.tags and decision == IMPORT: config_tag_names = set(config.tags.keys()) for tag in session.query(Tag): if tag.name not in config_tag_names: @@ -564,6 +572,7 @@ class TagsConfig(ConfigFile): if need_commit: session.commit() + self.record_import_version() self.successful_import = True logger.info('Importing tags config complete') except Exception as e: @@ -680,10 +689,13 @@ class TagsConfig(ConfigFile): config_file.parent.mkdir(parents=True, exist_ok=True) self.write_config_data(merged_data, config_file) - self.import_config(send_events=send_events) + # Importing a backup is the user's choice: apply it whatever the versions. + self.import_forced(send_events=send_events) TAGS_CONFIG: TagsConfig = TagsConfig() +register_ownership(TagsConfig.file_name, Tag) +register_ownership(TagsConfig.file_name, TagFile) TEST_TAGS_CONFIG: TagsConfig = None # noqa diff --git a/wrolpi/test/test_config_import_benchmark.py b/wrolpi/test/test_config_import_benchmark.py index af143a95..7c653eab 100644 --- a/wrolpi/test/test_config_import_benchmark.py +++ b/wrolpi/test/test_config_import_benchmark.py @@ -164,7 +164,8 @@ async def import_all(results: list, prefix: str): def wipe_config_rows(session): - for table in ('tag_file', 'collection_item', 'channel', 'repository', 'download', 'collection', 'tag'): + for table in ('tag_file', 'collection_item', 'channel', 'repository', 'download', 'collection', 'tag', + 'config_version'): session.execute(text(f'DELETE FROM {table}')) session.execute(text('UPDATE file_group SET viewed = NULL, progress = NULL, position = NULL')) session.commit() diff --git a/wrolpi/test/test_config_version.py b/wrolpi/test/test_config_version.py new file mode 100644 index 00000000..b979d4a4 --- /dev/null +++ b/wrolpi/test/test_config_version.py @@ -0,0 +1,336 @@ +"""A database config's file and the database each carry a version (see wrolpi.config_version).""" +import pytest +import yaml + +from wrolpi.collections.models import Collection +from wrolpi.config_version import get_config_version, set_config_version +from wrolpi.dates import now +from wrolpi.files.models import FileGroup +from wrolpi.tags import Tag, get_tags_config, import_tags_config + + +def write_tags_config(path, version: int, names): + path.write_text(yaml.dump(dict(version=version, tags={i: dict(color='#000000') for i in names}, + tag_files=[], tag_zims=[]))) + + +def db_version(session, name='tags.yaml'): + session.commit() + return get_config_version(session, name) + + +@pytest.mark.asyncio +async def test_owned_changes_count(test_session, test_directory, make_files_structure, async_client): + """Changing rows a config owns adds one to its database version; other changes do not.""" + set_config_version(test_session, 'tags.yaml', 5) + set_config_version(test_session, 'domains.yaml', 5) + set_config_version(test_session, 'channels.yaml', 5) + set_config_version(test_session, 'recently_viewed.yaml', 5) + test_session.commit() + + test_session.add(Tag(name='one', color='#000000')) + test_session.flush() + assert db_version(test_session) == 6 + + # A domain Collection belongs to domains.yaml, not channels.yaml. + test_session.add(Collection(name='example.com', kind='domain')) + test_session.flush() + assert db_version(test_session, 'domains.yaml') == 6 + assert db_version(test_session, 'channels.yaml') == 5 + + # recently_viewed.yaml owns only a FileGroup's viewing columns. + make_files_structure(['a.txt']) + file_group = FileGroup.from_paths(test_session, test_directory / 'a.txt') + test_session.add(file_group) + test_session.flush() + test_session.commit() + file_group.title = 'Retitled' + test_session.flush() + assert db_version(test_session, 'recently_viewed.yaml') == 5 + file_group.set_viewed() + test_session.flush() + assert db_version(test_session, 'recently_viewed.yaml') == 6 + + # A config without a database row is not created by a change; its next import creates it. + test_session.add(Collection(name='Mix', kind='playlist')) + test_session.flush() + assert db_version(test_session, 'playlists.yaml') is None + + +@pytest.mark.asyncio +async def test_dump_writes_database_version(test_session, test_tags_config, await_switches): + """A saved config carries the database's version, read with the rows it saves.""" + test_session.add(Tag(name='one', color='#000000')) + test_session.commit() + set_config_version(test_session, 'tags.yaml', 41) + test_session.commit() + + config = get_tags_config() + config.successful_import = True + config.dump_config() + await await_switches() + assert yaml.safe_load(test_tags_config.read_text())['version'] == 41 + + +@pytest.mark.asyncio +async def test_unsaved_database_change_is_saved_not_undone(test_session, test_tags_config, await_switches): + """WROLPi restarted after a Tag was created, before tags.yaml was saved: the import saves the database instead + of importing the older file (which would delete the new Tag).""" + test_session.add(Tag(name='old', color='#000000')) + test_session.commit() + write_tags_config(test_tags_config, 3, ['old']) + set_config_version(test_session, 'tags.yaml', 3) + test_session.add(Tag(name='new', color='#000000')) # Counted: the database is now version 4. + test_session.commit() + assert db_version(test_session) == 4 + + import_tags_config() + await await_switches() + + assert {i.name for i in test_session.query(Tag)} == {'old', 'new'} + saved = yaml.safe_load(test_tags_config.read_text()) + assert set(saved['tags']) == {'old', 'new'} and saved['version'] == 4 + + +@pytest.mark.asyncio +@pytest.mark.parametrize('file_version', [3, 9]) +async def test_current_file_imports_with_deletes(test_session, test_tags_config, file_version, async_client): + """A file as new as the database (or newer) is the truth: what it no longer lists is deleted. The database + records the file's version.""" + test_session.add_all([Tag(name='kept', color='#000000'), Tag(name='removed', color='#000000')]) + test_session.commit() + set_config_version(test_session, 'tags.yaml', 3) + test_session.commit() + write_tags_config(test_tags_config, file_version, ['kept']) + + import_tags_config() + + assert {i.name for i in test_session.query(Tag)} == {'kept'} + assert db_version(test_session) == file_version + + +@pytest.mark.asyncio +async def test_unknown_database_version_imports_without_deletes(test_session, test_tags_config, await_switches): + """A config this code never imported (no database row) imports without deleting. The database may keep rows + the file does not list, so the file's version is not recorded (the next import would delete them); they are + saved into the file instead, and that save records the version.""" + test_session.add_all([Tag(name='kept', color='#000000'), Tag(name='unlisted', color='#000000')]) + test_session.commit() + write_tags_config(test_tags_config, 7, ['kept', 'added']) + + import_tags_config() + assert {i.name for i in test_session.query(Tag)} == {'kept', 'unlisted', 'added'} + assert db_version(test_session) is None + + await await_switches() + saved = yaml.safe_load(test_tags_config.read_text()) + assert set(saved['tags']) == {'kept', 'unlisted', 'added'} and saved['version'] == 8 + assert db_version(test_session) == 8 + + +@pytest.mark.asyncio +async def test_failed_import_leaves_version_unknown(test_session, test_tags_config, monkeypatch, async_client): + """An import's own writes do not count: one that fails partway leaves the version unknown, not newer than the + file it did not finish applying (which would save that partial database over the file).""" + from wrolpi.tags import TagsConfig + + test_session.add(Tag(name='kept', color='#000000')) + test_session.commit() + set_config_version(test_session, 'tags.yaml', 3) + test_session.commit() + write_tags_config(test_tags_config, 3, ['kept', 'added']) + + def fail(self): + raise RuntimeError('Import failed') + + monkeypatch.setattr(TagsConfig, 'record_import_version', fail) + with pytest.raises(RuntimeError): + import_tags_config() + + assert {i.name for i in test_session.query(Tag)} == {'kept', 'added'} + assert db_version(test_session) is None + + +@pytest.mark.asyncio +async def test_import_button_imports_an_older_file(test_session, test_tags_config, async_client): + """The Import button applies the file whatever its version: the user chose it.""" + test_session.add_all([Tag(name='kept', color='#000000'), Tag(name='removed', color='#000000')]) + test_session.commit() + set_config_version(test_session, 'tags.yaml', 10) + test_session.commit() + write_tags_config(test_tags_config, 2, ['kept']) + + get_tags_config().import_forced() + + assert {i.name for i in test_session.query(Tag)} == {'kept'} + assert db_version(test_session) == 2 + + +@pytest.mark.asyncio +async def test_raw_writes_count(test_session, test_directory, make_files_structure, async_client): + """Writes that bypass the ORM flush count their changes explicitly.""" + from wrolpi.downloader import Download, DownloadManager, DownloadStatus + from wrolpi.files.lib import _bulk_delete_tag_files, _bulk_insert_tag_files + + make_files_structure(['a.txt']) + file_group = FileGroup.from_paths(test_session, test_directory / 'a.txt') + tag = Tag(name='one', color='#000000') + test_session.add_all([file_group, tag]) + test_session.commit() + set_config_version(test_session, 'tags.yaml', 1) + test_session.commit() + + _bulk_insert_tag_files([(tag.id, file_group.id)]) + assert db_version(test_session) == 2 + _bulk_delete_tag_files([(tag.id, file_group.id)]) + assert db_version(test_session) == 3 + + # Untagging deletes in bulk. + file_group.add_tag(test_session, tag.id) + test_session.commit() + assert db_version(test_session) == 4 + file_group.untag(test_session, tag.id) + test_session.commit() + assert db_version(test_session) == 5 + + # Retrying Downloads updates their status in bulk. + test_session.add(Download(url='https://example.com/1', status=DownloadStatus.failed)) + test_session.commit() + set_config_version(test_session, 'download_manager.yaml', 1) + test_session.commit() + DownloadManager.retry_downloads(test_session) + test_session.commit() + assert db_version(test_session, 'download_manager.yaml') == 2 + + +def test_downloads_count_for_their_collections_config_only(test_session): + """A Channel's Download is saved by channels.yaml, not repos.yaml: counting it for repos.yaml would make the repos + import save instead of import.""" + from wrolpi.downloader import Download + + channel = Collection(name='Channel', kind='channel') + test_session.add(channel) + test_session.commit() + for name in ('channels.yaml', 'repos.yaml', 'download_manager.yaml'): + set_config_version(test_session, name, 1) + test_session.commit() + + test_session.add(Download(url='https://example.com/channel', frequency=86400, collection_id=channel.id)) + test_session.commit() + + assert db_version(test_session, 'channels.yaml') == 2 + assert db_version(test_session, 'download_manager.yaml') == 2 + assert db_version(test_session, 'repos.yaml') == 1 + + +def test_download_leaving_a_collection_counts_for_its_config(test_session): + """Moving a Download out of a Channel changes channels.yaml: otherwise its next import would put it back.""" + from wrolpi.downloader import Download + + channel = Collection(name='Channel', kind='channel') + test_session.add(channel) + test_session.commit() + download = Download(url='https://example.com/channel', frequency=86400, collection_id=channel.id) + test_session.add(download) + test_session.commit() + set_config_version(test_session, 'channels.yaml', 1) + test_session.commit() + + download.collection_id = None + test_session.commit() + + assert db_version(test_session, 'channels.yaml') == 2 + + +def test_deleting_once_downloads_counts_for_download_manager_only(test_session, test_download_manager): + """Deleting a once-download (no Collection) does not change channels.yaml or repos.yaml: counting it would make + their next import save instead of import.""" + from wrolpi.downloader import Download + + download = Download(url='https://example.com/once') + test_session.add(download) + test_session.commit() + for name in ('channels.yaml', 'repos.yaml', 'download_manager.yaml'): + set_config_version(test_session, name, 1) + test_session.commit() + + test_download_manager.delete_downloads_by_ids(test_session, [download.id]) + + assert db_version(test_session, 'download_manager.yaml') == 2 + assert db_version(test_session, 'channels.yaml') == 1 + assert db_version(test_session, 'repos.yaml') == 1 + + +def test_only_playlist_items_count_for_playlists(test_session): + """Author and subject Collections have items too; only a playlist's are saved by playlists.yaml.""" + author = Collection(name='Author', kind='author') + playlist = Collection(name='Mix', kind='playlist') + test_session.add_all([author, playlist]) + test_session.commit() + set_config_version(test_session, 'playlists.yaml', 1) + test_session.commit() + + author.add_url(test_session, '/map?x=1', title='Spot') + test_session.commit() + assert db_version(test_session, 'playlists.yaml') == 1 + + playlist.add_url(test_session, '/map?x=1', title='Spot') + test_session.commit() + assert db_version(test_session, 'playlists.yaml') == 2 + + +def test_unsaved_columns_do_not_count(test_session): + """Columns a config does not save (a fetch's commit, a refresh's counts, a download attempt) do not count: + counting them would make that config's next import save instead of import.""" + from modules.repos.models import Repository + from modules.videos.models import Channel + from wrolpi.downloader import Download + + channel_collection = Collection(name='Channel', kind='channel') + repo_collection = Collection(name='Repo', kind='repo') + test_session.add_all([channel_collection, repo_collection]) + test_session.flush() + channel = Channel(collection_id=channel_collection.id) + repo = Repository(url='https://example.com/repo.git', collection_id=repo_collection.id) + download = Download(url='https://example.com/channel', frequency=86400, collection_id=channel_collection.id) + test_session.add_all([channel, repo, download]) + test_session.commit() + names = ('channels.yaml', 'repos.yaml', 'download_manager.yaml') + for name in names: + set_config_version(test_session, name, 1) + test_session.commit() + + repo.head_sha = 'abc123' + repo.last_fetch = now() + channel.video_count = 10 + channel.refreshed = True + channel_collection.item_count = 10 + repo_collection.total_size = 1000 + download.attempts = 1 + download.error = 'Failed' + download.last_download_attempt = now() + test_session.commit() + assert {i: db_version(test_session, i) for i in names} == dict.fromkeys(names, 1) + + # Saved columns still count. + repo.branch = 'dev' + channel.generate_posters = True + download.status = 'failed' + test_session.commit() + assert {i: db_version(test_session, i) for i in names} == dict.fromkeys(names, 2) + + +def test_domain_downloads_count_for_domains(test_session): + """domains.yaml saves each domain's Downloads (url, frequency).""" + from wrolpi.downloader import Download + + domain = Collection(name='example.com', kind='domain') + test_session.add(domain) + test_session.commit() + set_config_version(test_session, 'domains.yaml', 1) + test_session.commit() + + test_session.add(Download(url='https://example.com/feed', frequency=86400, collection_id=domain.id)) + test_session.commit() + + assert db_version(test_session, 'domains.yaml') == 2 diff --git a/wrolpi/test/test_db.py b/wrolpi/test/test_db.py index 2d071f87..5ae9e2d7 100644 --- a/wrolpi/test/test_db.py +++ b/wrolpi/test/test_db.py @@ -360,7 +360,9 @@ def test_deferred_request_writes_are_recorded(test_session): db.session.flush() db.finish(ok=False) - statements = [statement for statement, _ in db_module.DEFERRED_REQUEST_WRITES] + # A Tag is saved by tags.yaml, so its flush also counts the change (wrolpi.config_version). + statements = [statement for statement, _ in db_module.DEFERRED_REQUEST_WRITES + if 'config_version' not in statement] db_module.DEFERRED_REQUEST_WRITES.clear() assert len(statements) == 1 and statements[0].startswith('INSERT INTO tag'), statements @@ -390,6 +392,8 @@ async def test_task_started_in_a_write_does_not_inherit_its_intent(test_session) await task db.finish(ok=True) - statements = [statement for statement, _ in db_module.DEFERRED_REQUEST_WRITES] + # A Tag is saved by tags.yaml, so its flush also counts the change (wrolpi.config_version). + statements = [statement for statement, _ in db_module.DEFERRED_REQUEST_WRITES + if 'config_version' not in statement] db_module.DEFERRED_REQUEST_WRITES.clear() assert len(statements) == 1 and statements[0].startswith('INSERT INTO tag'), statements diff --git a/wrolpi/test/test_downloads_config.py b/wrolpi/test/test_downloads_config.py index fba264d5..98ad5517 100644 --- a/wrolpi/test/test_downloads_config.py +++ b/wrolpi/test/test_downloads_config.py @@ -9,6 +9,7 @@ Tests cover: import pytest import yaml +from wrolpi.conftest import config_was_imported from wrolpi.downloader import Download, get_download_manager_config @@ -242,6 +243,7 @@ class TestDownloadManagerConfigEdgeCases: # Reinitialize and import config.initialize() + config_was_imported(test_session, 'download_manager.yaml') config.import_config() assert config.successful_import is True @@ -425,6 +427,7 @@ async def test_import_config_preserves_completed_once_downloads(test_session, te yaml.dump(config_data, f) config.initialize() + config_was_imported(test_session, 'download_manager.yaml') config.import_config() assert config.successful_import is True diff --git a/wrolpi/test/test_tags.py b/wrolpi/test/test_tags.py index 32affa09..e34f4660 100644 --- a/wrolpi/test/test_tags.py +++ b/wrolpi/test/test_tags.py @@ -6,6 +6,7 @@ import yaml from wrolpi import tags from wrolpi.common import is_hardlinked, walk, get_wrolpi_config +from wrolpi.conftest import config_was_imported from wrolpi.errors import FileGroupIsTagged, InvalidTag, UnknownTag, UsedTag from wrolpi.files.models import FileGroup from wrolpi.tags import TagFile, Tag @@ -382,6 +383,7 @@ async def test_import_tags_delete_missing(test_session, test_directory, test_tag assert len(config.tags) == 1 # Import the config. Only unused Tags not in the config are deleted. + config_was_imported(test_session, 'tags.yaml') tags.import_tags_config() assert {i.name for i in test_session.query(tags.Tag)} == { tag1.name, # tag1 is used by FileGroup.