From deba43366a52317ac8eb424dfde694cc6f1e9e2e Mon Sep 17 00:00:00 2001 From: Roland Date: Sun, 25 Oct 2026 09:54:53 -0600 Subject: [PATCH] repos.yaml imports by diffing repos.yaml declares one section, its Repos keyed by repo_url_key (another URL of the same Repo is that Repo), comparing name, description, directory, Tag, mode, branch, submodules, frequency and clone token (a missing directory or clone token keeps the database's). An unchanged import writes nothing (every import rewrote every Repo and its Download). An invalid entry is skipped and counted, and its Repo is kept; before, a Repo whose entry failed validation was deleted. New Repos are created inside the import's one write: recurring_download (via get_or_create_download and _set_repository_download) takes commit=False, and the URL is not looked up again for each new Repo (O(N^2)). Co-Authored-By: Claude Opus 5.5 --- modules/repos/lib.py | 221 ++++++++++++----------- modules/repos/test/test_config_import.py | 119 ++++++++++++ modules/repos/test/test_lib.py | 8 +- modules/repos/test/test_search.py | 4 +- wrolpi/collections/models.py | 5 +- wrolpi/downloader.py | 9 +- 6 files changed, 253 insertions(+), 113 deletions(-) create mode 100644 modules/repos/test/test_config_import.py diff --git a/modules/repos/lib.py b/modules/repos/lib.py index b56ef585..f25992a2 100644 --- a/modules/repos/lib.py +++ b/modules/repos/lib.py @@ -14,10 +14,10 @@ import subprocess import tempfile from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone -from typing import Callable, ContextManager, List, NamedTuple, Optional, Tuple +from typing import Callable, ContextManager, Dict, List, NamedTuple, Optional, Set, Tuple from urllib.parse import urlparse -from sqlalchemy import func, text +from sqlalchemy import func, select, text from sqlalchemy.orm import Session from wrolpi import fts @@ -25,7 +25,8 @@ 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.config_import import ConfigSection, media_relative +from wrolpi.config_version import 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 @@ -45,7 +46,7 @@ __all__ = ['ParsedRepoURL', 'parse_repo_url', 'repo_url_key', 'git_command', 'gi 'find_readme', 'repos_destination_root', 'validate_repos_destination', 'is_valid_branch', 'submodule_url_error', 'HostAddresses', 'resolve_repo_addresses', 'pin_config', 'is_public_address', 'submodule_fetch_config', 'read_readme_text', 'search_repos', 'count_repos', 'search_repos_by_name', - 'get_repo_log', 'get_repo_root', 'repo_archive_name', 'stream_repo_archive', 'ReposConfig', + 'get_repo_log', 'get_repo_root', 'repo_archive_name', 'stream_repo_archive', 'ReposConfig', 'ReposSection', 'repos_config', 'get_repos_config', 'save_repos_config', 'import_repos_config', 'DEFAULT_REPO_FREQUENCY'] DEFAULT_REPO_FREQUENCY = DownloadFrequency.weekly.value @@ -734,9 +735,11 @@ def create_repository(session: Session, url: str, tag_name: str = None, frequenc return repo -def _set_repository_download(session: Session, repo: Repository, frequency: int, clone: bool = False): +def _set_repository_download(session: Session, repo: Repository, frequency: int, clone: bool = False, + commit: bool = True): """Create (or change) the recurring Download which keeps the Repo up to date. A Repo which is never updated - (`frequency` is NEVER) has none; with `clone`, it is cloned once if its clone is missing.""" + (`frequency` is NEVER) has none; with `clone`, it is cloned once if its clone is missing. `commit`: False + leaves committing to the caller's write.""" if not frequency: if download := repo.download: # A deleted object stays in loaded collections; remove it so the Repo reads as never updated. @@ -747,7 +750,8 @@ def _set_repository_download(session: Session, repo: Repository, frequency: int, return once = repo.once_download - download = repo.collection.get_or_create_download(session, repo.url, frequency, downloader_name='git') + download = repo.collection.get_or_create_download(session, repo.url, frequency, downloader_name='git', + commit=commit) # The Download may already exist (e.g. imported from download_manager.yaml, or the Repo's one-time Download). download.frequency = frequency if once: @@ -766,8 +770,9 @@ def _create_repository_download(session: Session, repo: Repository, frequency: i def _create_repository_records(session: Session, url: str, tag_name: str, mode: str, branch: Optional[str], name: Optional[str], description: Optional[str], directory: Optional[pathlib.Path], - submodules: bool, clone_token: str = None) -> Repository: - """Create (but do not commit) a Repo's Collection and Repository, in their directory.""" + submodules: bool, clone_token: str = None, check_url: bool = True) -> Repository: + """Create (but do not commit) a Repo's Collection and Repository, in their directory. `check_url`: refuse a URL + another Repo has (the config import already knows it is new).""" parsed = parse_repo_url(url) _validate_mode(mode) branch = branch or parsed.branch or None @@ -776,7 +781,7 @@ def _create_repository_records(session: Session, url: str, tag_name: str, mode: if name in ('', '.', '..'): raise InvalidRepo(f'Invalid repo name: {name!r}') - if get_repository_by_url(session, parsed.url): + if check_url and get_repository_by_url(session, parsed.url): raise RepoConflict(f'A repo with this URL already exists: {parsed.url}') if session.query(Collection).filter_by(kind='repo', name=name).one_or_none(): suggestion = f'{parsed.owner.replace("/", "-")}-{name}' if parsed.owner else f'{parsed.host}-{name}' @@ -1368,6 +1373,107 @@ def get_repo_size(directory: pathlib.Path) -> int: return total +def _repo_row(url, name, description, directory, tag_name, mode, branch, submodules, frequency, clone_token) -> dict: + return dict( + url=url, + name=name, + description=description or None, + directory=media_relative(directory) if directory else None, + tag_name=tag_name or None, + mode=mode, + branch=branch or None, + submodules=bool(submodules), + frequency=frequency or NEVER, + clone_token=clone_token or None, + ) + + +class ReposSection(ConfigSection): + """repos.yaml's Repos (a Collection with kind 'repo', its Repository, and its recurring Download), keyed by + `repo_url_key`, so another URL of the same Repo is that Repo. An invalid entry is skipped, and kept.""" + name = 'repos' + + def file_rows(self, config: dict) -> Tuple[Dict[str, dict], Set[str]]: + media_directory = get_media_directory() + rows, invalid = dict(), set() + for position, data in enumerate(config.get('repos') or []): + url = data.get('url') + if not url: + logger.error(f'Repo config entry has no url, skipping: {data}') + # Counted as skipped; never a Repo's key. + invalid.add(f'\0{position}') + continue + key = repo_url_key(url) + try: + if key in rows: + raise InvalidRepo(f'Another entry has the same repo: {url}') + parsed = parse_repo_url(url) + mode = data.get('mode') or 'full' + _validate_mode(mode) + frequency = data.get('frequency') + frequency = NEVER if frequency in (NEVER_CONFIG, NEVER) else (frequency or DEFAULT_REPO_FREQUENCY) + _validate_frequency(frequency) + directory = media_directory / data['directory'] if data.get('directory') else None + if directory: + _validate_directory(directory) + branch = data.get('branch') or parsed.branch or None + _validate_branch(branch) + rows[key] = _repo_row(parsed.url, data.get('name') or parsed.name, data.get('description'), + directory, data.get('tag_name'), mode, branch, data.get('submodules'), + frequency, data.get('clone_token')) + except Exception as e: + logger.error(f'Skipping invalid repo in config: {data}', exc_info=e) + invalid.add(key) + return rows, invalid + + def db_rows(self, session: Session) -> Dict[str, dict]: + # The recurring Download (see `Repository.download`). + frequency = select(Download.frequency) \ + .where(Download.collection_id == Collection.id, Download.frequency > 0) \ + .limit(1).scalar_subquery() + query = session.query(Repository.url, Collection.name, Collection.description, Collection.directory, + Tag.name, Repository.mode, Repository.branch, Repository.submodules, frequency, + Repository.clone_token) \ + .join(Collection, Collection.id == Repository.collection_id) \ + .outerjoin(Tag, Tag.id == Collection.tag_id) + return {repo_url_key(url): _repo_row(url, *values) for url, *values in query} + + def same(self, file_row: dict, db_row: dict) -> bool: + # The key is the URL. A file without a directory or clone token keeps the database's. + return all(value == db_row[name] for name, value in file_row.items() + if name != 'url' and not (name in ('directory', 'clone_token') and value is None)) + + def upsert(self, session: Session, rows: Dict[str, dict]): + media_directory = get_media_directory() + existing = {repo_url_key(i.url): i for i in session.query(Repository)} + for key, row in rows.items(): + directory = media_directory / row['directory'] if row['directory'] else None + if not (repo := existing.get(key)): + repo = _create_repository_records(session, row['url'], row['tag_name'], row['mode'], row['branch'], + row['name'], row['description'], directory, row['submodules'], + row['clone_token'], check_url=False) + _set_repository_download(session, repo, row['frequency'], clone=True, commit=False) + continue + collection = repo.collection + collection.name = row['name'] + repo.update_search_name() + collection.description = row['description'] + collection.tag = Tag.get_or_create_tag(session, row['tag_name']) if row['tag_name'] else None + if directory: + collection.directory = directory + repo.mode, repo.branch, repo.submodules = row['mode'], row['branch'], row['submodules'] + if row['clone_token']: + repo.clone_token = row['clone_token'] + _set_repository_download(session, repo, row['frequency'], commit=False) + session.flush() + + def delete(self, session: Session, keys: Set[str]): + for repo in session.query(Repository): + if repo_url_key(repo.url) in keys: + logger.info(f'Deleting repo {repo.name!r} (no longer in config)') + _delete_repository(session, repo) + + @dataclass class ReposConfigValidator: version: int = 0 @@ -1389,6 +1495,7 @@ class ReposConfig(ConfigFile): frequency: 604800 """ file_name = 'repos.yaml' + sections = (ReposSection(),) validator = ReposConfigValidator default_config = dict(version=0, repos=[]) width = 120 @@ -1397,100 +1504,6 @@ class ReposConfig(ConfigFile): def repos(self) -> List[dict]: return self._config.get('repos', []) - def import_config(self, file: pathlib.Path = None, send_events=False): - """Import Repos from the config into the database.""" - file = file or self.get_file() - file_str = str(self.get_relative_file()) - - # A missing config is not an empty config; never delete Repos because a file is lost. - if not file.is_file(): - logger.info(f'No repos config file, skipping import: {file_str}') - self.successful_import = True - 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. - if not repos_data: - logger.info(f'No repos in config, preserving existing DB repos: {file_str}') - self.successful_import = True - return - - try: - with get_db_session(commit=True) as session: - imported_ids = set() - for data in repos_data: - try: - repo = self._import_repo(session, data) - except Exception as e: - logger.error(f'Failed to import repo from config: {data}', exc_info=e) - self.import_skipped += 1 - continue - imported_ids.add(repo.id) - - 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 - message = f'Failed to import {file_str} config!' - logger.error(message, exc_info=e) - if send_events: - Events.send_config_import_failed(message) - raise - - @staticmethod - def _import_repo(session: Session, data: dict) -> Repository: - parsed = parse_repo_url(data['url']) - name = data.get('name') or parsed.name - mode = data.get('mode') or 'full' - _validate_mode(mode) - frequency = data.get('frequency') - frequency = NEVER if frequency in (NEVER_CONFIG, NEVER) else (frequency or DEFAULT_REPO_FREQUENCY) - _validate_frequency(frequency) - directory = data.get('directory') - directory = get_media_directory() / directory if directory else None - if directory: - _validate_directory(directory) - tag_name = data.get('tag_name') - - repo = get_repository_by_url(session, parsed.url) - if not repo: - return create_repository(session, parsed.url, tag_name=tag_name, frequency=frequency, mode=mode, - branch=data.get('branch'), name=name, description=data.get('description'), - directory=directory, submodules=bool(data.get('submodules')), - clone_token=data.get('clone_token')) - - collection = repo.collection - collection.name = name - repo.update_search_name() - collection.description = data.get('description') - collection.tag = Tag.get_or_create_tag(session, tag_name) if tag_name else None - if directory: - collection.directory = directory - repo.mode = mode - repo.submodules = bool(data.get('submodules')) - if data.get('clone_token'): - repo.clone_token = data['clone_token'] - branch = data.get('branch') or None - _validate_branch(branch) - repo.branch = branch - _set_repository_download(session, repo, frequency) - session.flush() - return repo - 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: diff --git a/modules/repos/test/test_config_import.py b/modules/repos/test/test_config_import.py new file mode 100644 index 00000000..8ff712ed --- /dev/null +++ b/modules/repos/test/test_config_import.py @@ -0,0 +1,119 @@ +"""repos.yaml imports by diffing its Repos against the database, and writing only the differences (see +wrolpi.config_import). A Repo is keyed by its URL (`repo_url_key`).""" +import pytest +import yaml + +from modules.repos import lib +from modules.repos.models import Repository +from wrolpi.config_version import get_config_version +from wrolpi.downloader import DownloadFrequency +from wrolpi.test.test_config_import import edit_config, statements, writes + +ONE = 'https://github.com/a/one' +TWO = 'https://github.com/a/two' + + +@pytest.fixture +async def saved_repos(test_session, async_client, test_wrolpi_config, await_switches): + """Repo `one` (tagged, daily) and `two` (a snapshot of a branch), saved to the config (which records its + version).""" + lib.create_repository(test_session, ONE, tag_name='software', description='One', frequency=DownloadFrequency.daily) + lib.create_repository(test_session, TWO, mode='snapshot', branch='dev') + await await_switches() + config = lib.get_repos_config() + config.dump_config() + await await_switches() + assert get_config_version(test_session, 'repos.yaml') is not None + return config.get_file() + + +def edit_repo(path, repo_url, **values): + def change(data): + next(i for i in data["repos"] if i["url"] == repo_url).update(values) + + edit_config(path, change) + + +def repos(test_session) -> dict: + test_session.expire_all() + return {i.url: i for i in test_session.query(Repository)} + + +@pytest.mark.asyncio +async def test_unchanged_import_writes_nothing(test_session, saved_repos): + """Importing a config the database already matches opens no write transaction.""" + config = lib.get_repos_config() + with statements() as collected: + config.import_config() + + assert config.successful_import is True + assert writes(collected) == [] + # The version, then one snapshot of the Repos. + assert len(collected) <= 3, collected + + +@pytest.mark.asyncio +async def test_same_repo_by_another_url_is_unchanged(test_session, saved_repos): + """A URL of the same Repo (a `.git` suffix, the host's case) is that Repo.""" + edit_repo(saved_repos, TWO, url='https://GitHub.com/a/two.git') + config = lib.get_repos_config() + with statements() as collected: + config.import_config() + + assert writes(collected) == [] + assert set(repos(test_session)) == {ONE, TWO} + + +@pytest.mark.asyncio +async def test_import_writes_only_the_differences(test_session, saved_repos): + """A changed Repo is updated, a removed one is deleted (with its Download), an added one is created.""" + def change(data): + one = next(i for i in data['repos'] if i['url'] == ONE) + one.update(description='Changed', frequency='never', mode='snapshot', branch='main') + data['repos'] = [one, dict(url='https://github.com/a/three', mode='full')] + + edit_config(saved_repos, change) + config = lib.get_repos_config() + config.import_config() + + assert config.successful_import is True + found = repos(test_session) + assert set(found) == {ONE, 'https://github.com/a/three'} + one = found[ONE] + assert (one.collection.description, one.mode, one.branch, one.download) == ('Changed', 'snapshot', 'main', None) + assert found['https://github.com/a/three'].download.frequency == lib.DEFAULT_REPO_FREQUENCY + with statements() as collected: + config.import_config() + assert writes(collected) == [] + + +@pytest.mark.asyncio +async def test_invalid_entry_is_kept(test_session, saved_repos): + """A Repo whose entry is invalid is skipped, not deleted; the others are still imported.""" + edit_repo(saved_repos, TWO, mode='bogus') + edit_repo(saved_repos, ONE, description='Changed') + + config = lib.get_repos_config() + config.import_config() + + assert config.import_skipped == 1 + found = repos(test_session) + assert set(found) == {ONE, TWO} + assert (found[TWO].mode, found[ONE].collection.description) == ('snapshot', 'Changed') + + +@pytest.mark.asyncio +async def test_failed_import_creates_nothing(test_session, saved_repos, monkeypatch): + """A new Repo is created in the import's one write: a failure later in the write leaves nothing (each new Repo + was committed on its own).""" + edit_config(saved_repos, lambda data: data['repos'].append(dict(url='https://github.com/a/three'))) + edit_config(saved_repos, lambda data: data.update(repos=[i for i in data['repos'] if i['url'] != TWO])) + + def fail(*args, **kwargs): + raise RuntimeError('Write failed') + + monkeypatch.setattr(lib, '_delete_repository', fail) + with pytest.raises(RuntimeError): + lib.get_repos_config().import_config() + + assert set(repos(test_session)) == {ONE, TWO} diff --git a/modules/repos/test/test_lib.py b/modules/repos/test/test_lib.py index 454bb520..46859a76 100644 --- a/modules/repos/test/test_lib.py +++ b/modules/repos/test/test_lib.py @@ -277,11 +277,11 @@ async def test_repo_directory_traversal(test_session, test_directory, async_clie lib.create_repository(test_session, 'https://github.com/a/b', directory=test_directory) test_session.rollback() - # The config can not move an existing Repo outside the media directory. + # The config can not move an existing Repo outside the media directory: the entry is invalid (and skipped). lib.create_repository(test_session, 'https://github.com/a/b') - with pytest.raises(InvalidRepo): - lib.ReposConfig._import_repo(test_session, dict(url='https://github.com/a/b', directory='../outside')) - test_session.rollback() + entry = dict(url='https://github.com/a/b', directory='../outside') + rows, invalid = lib.ReposSection().file_rows(dict(repos=[entry])) + assert (rows, invalid) == (dict(), {lib.repo_url_key('https://github.com/a/b')}) assert test_session.query(Repository).one().directory == test_directory / 'repos/b' diff --git a/modules/repos/test/test_search.py b/modules/repos/test/test_search.py index 9ff59f0d..78c5ac34 100644 --- a/modules/repos/test/test_search.py +++ b/modules/repos/test/test_search.py @@ -63,7 +63,9 @@ async def test_search_index_follows_changes(test_session, test_directory, async_ assert lib.count_repos(test_session, 'first') == 0 assert lib.count_repos(test_session, 'second') == 1 - lib.ReposConfig._import_repo(test_session, dict(url='https://github.com/a/b', name='renamed')) + section = lib.ReposSection() + rows, _ = section.file_rows(dict(repos=[dict(url='https://github.com/a/b', name='renamed')])) + section.upsert(test_session, rows) test_session.commit() assert lib.count_repos(test_session, 'renamed') == 1 diff --git a/wrolpi/collections/models.py b/wrolpi/collections/models.py index 047cb35c..d43ed432 100644 --- a/wrolpi/collections/models.py +++ b/wrolpi/collections/models.py @@ -250,7 +250,8 @@ class Collection(ModelHelper, Base): def get_or_create_download(self, session: Session, url: str, frequency: int, reset_attempts: bool = False, downloader_name: str = None, - sub_downloader_name: str = None) -> 'Download': + sub_downloader_name: str = None, + commit: bool = True) -> 'Download': """Get or create a Download for this Collection. Args: @@ -260,6 +261,7 @@ class Collection(ModelHelper, Base): reset_attempts: Reset download attempts if True downloader_name: Name of the downloader to use sub_downloader_name: Name of the sub-downloader for child downloads + commit: Commit a new Download (False: the caller's write commits) Returns: The Download instance @@ -283,6 +285,7 @@ class Collection(ModelHelper, Base): sub_downloader_name=sub_downloader_name, destination=str(self.directory) if self.directory else None, reset_attempts=reset_attempts, + commit=commit, ) if reset_attempts: download.attempts = 0 diff --git a/wrolpi/downloader.py b/wrolpi/downloader.py index 49523175..559c7ec6 100644 --- a/wrolpi/downloader.py +++ b/wrolpi/downloader.py @@ -1043,8 +1043,8 @@ class DownloadManager: sub_downloader_name: str = None, reset_attempts: bool = False, destination: str | pathlib.Path = None, tag_names: List[str] = None, settings: Dict = None, collection_id: int = None, - override_skip: bool = False) -> Download: - """Schedule a recurring download.""" + override_skip: bool = False, commit: bool = True) -> Download: + """Schedule a recurring download. `commit`: False only flushes (the caller's write commits).""" if not frequency or not isinstance(frequency, int): raise ValueError('Recurring download must have a frequency!') @@ -1065,7 +1065,10 @@ class DownloadManager: if channel := Channel.get_by_url(session, url=download.url): download.collection_id = channel.collection_id - session.commit() + if commit: + session.commit() + else: + session.flush([download]) return download