From d77e69f54e4884ae2f9121dd3dbe4174297e1dea Mon Sep 17 00:00:00 2001 From: Roland Date: Thu, 8 Dec 2022 13:46:19 -0700 Subject: [PATCH] Adding refresh_files_list function and api endpoint which allows refreshing of specific files. --- wrolpi/files/api.py | 15 ++++++++++ wrolpi/files/lib.py | 53 ++++++++++++++++++++++++++++++++++- wrolpi/files/schema.py | 6 ++++ wrolpi/files/test/test_api.py | 11 ++++++++ wrolpi/files/test/test_lib.py | 40 ++++++++++++++++++++++++++ 5 files changed, 124 insertions(+), 1 deletion(-) diff --git a/wrolpi/files/api.py b/wrolpi/files/api.py index 65101625..239bff60 100644 --- a/wrolpi/files/api.py +++ b/wrolpi/files/api.py @@ -93,6 +93,21 @@ async def refresh_directory(_: Request, body: schema.DirectoryRefreshRequest): return response.empty() +@bp.post('/refresh/list') +@openapi.definition( + summary='Index all files in the provided list', + body=schema.FilesRefreshListRequest, +) +@validate(schema.FilesRefreshListRequest) +@wrol_mode_check +async def refresh_files_list(_: Request, body: schema.FilesRefreshListRequest): + if PYTEST: + await lib.refresh_files_list(body.files, body.include_files_near) + else: + background_task(lib.refresh_files_list(body.files, body.include_files_near)) + return response.empty() + + @bp.post('/search') @openapi.definition( summary='Search Files', diff --git a/wrolpi/files/lib.py b/wrolpi/files/lib.py index cdd85565..91105548 100644 --- a/wrolpi/files/lib.py +++ b/wrolpi/files/lib.py @@ -1,5 +1,6 @@ import asyncio import datetime +import functools import glob import os import pathlib @@ -16,7 +17,7 @@ from sqlalchemy.orm import Session from wrolpi.cmd import which from wrolpi.common import get_media_directory, wrol_mode_check, logger, limit_concurrent, \ get_files_and_directories, apply_modelers, apply_after_refresh, get_model_by_table_name, chunks_by_name, \ - background_task + background_task, partition from wrolpi.dates import now from wrolpi.db import get_db_session, get_db_curs, get_ranked_models from wrolpi.errors import InvalidFile @@ -118,6 +119,7 @@ SUFFIXES = { } +@functools.lru_cache() def split_path_stem_and_suffix(path: Union[pathlib.Path, str]) -> Tuple[str, str]: """Get the path's stem and suffix. @@ -211,6 +213,55 @@ def cancelable_wrapper(func: callable): return wrapped +@wrol_mode_check +async def refresh_files_list(paths: List[str], include_files_near: bool = True): + """Refresh a list of files. Will not refresh directories or subdirectories. + + Parameters: + paths: A list of files that need to be refreshed. If not absolute, they are assumed to be relative to the + media directory. + include_files_near: If true, will also refresh files that share the stem of the provided files. + """ + refresh_logger.info(f'Refreshing {len(paths)} files') + + media_directory = get_media_directory() + + # Convert all paths to absolute paths in the media directory. + paths = [pathlib.Path(i) for i in paths] + paths = [media_directory / i if not i.is_absolute() else i for i in paths] + + if not paths: + raise FileNotFoundError('No files to refresh') + + if include_files_near: + new_paths = [] + for path in paths: + new_paths.extend(glob_shared_stem(path)) + paths.extend(new_paths) + + paths, deleted_paths = partition(lambda i: i.exists(), paths) + + idempotency = now() + await _refresh_files_list(paths, idempotency) + + # Delete any files near the paths that no longer exist. + paths_stems = [split_path_stem_and_suffix(i)[0] for i in paths] + paths_stems = [str(media_directory / i) for i in paths_stems] + with get_db_curs(commit=True) as curs: + # Get all paths that match the stems of these refreshed files. + params = dict(paths_stems=paths_stems, idempotency=idempotency) + curs.execute(''' + UPDATE file SET idempotency = %(idempotency)s + WHERE full_stem = ANY (%(paths_stems)s) + RETURNING path + ''', params) + existing_paths = [pathlib.Path(i[0]) for i in curs.fetchall()] + # Delete any files which no longer exist. + deleted_paths.extend([i for i in existing_paths if not i.exists()]) + deleted_paths = list(map(str, deleted_paths)) + curs.execute('DELETE FROM file WHERE path = ANY(%(deleted_paths)s)', dict(deleted_paths=deleted_paths)) + + @limit_concurrent(1) # Only one refresh at a time. @wrol_mode_check @cancelable_wrapper diff --git a/wrolpi/files/schema.py b/wrolpi/files/schema.py index a3899c33..93a3adac 100644 --- a/wrolpi/files/schema.py +++ b/wrolpi/files/schema.py @@ -26,6 +26,12 @@ class DirectoryRefreshRequest: directory: str = None +@dataclass +class FilesRefreshListRequest: + files: List[str] = None + include_files_near: bool = True + + @dataclass class DirectoriesRequest: search_str: Optional[str] = '' diff --git a/wrolpi/files/test/test_api.py b/wrolpi/files/test/test_api.py index 328f2e40..b9158e5c 100644 --- a/wrolpi/files/test/test_api.py +++ b/wrolpi/files/test/test_api.py @@ -231,3 +231,14 @@ def test_directory_search(test_client, make_files_structure): assert_directories('foo', ['foo']) # Does not exist. assert_directories('food', []) + + +def test_refresh_files_list(test_session, test_client, make_files_structure): + """The user can request to refresh specific files.""" + make_files_structure(['bar.txt', 'bar.mp4']) + + # Video file near `bar.txt` can be ignored. + content = json.dumps({'files': ['bar.txt'], 'include_files_near': False}) + request, response = test_client.post('/api/files/refresh/list', content=content) + assert response.status_code == HTTPStatus.NO_CONTENT + assert test_session.query(File).count() == 1 diff --git a/wrolpi/files/test/test_lib.py b/wrolpi/files/test/test_lib.py index f9f0dce1..5deb9886 100644 --- a/wrolpi/files/test/test_lib.py +++ b/wrolpi/files/test/test_lib.py @@ -199,6 +199,46 @@ async def test_refresh_many_files(test_session, make_files_structure): assert test_session.query(File).count() == count +@pytest.mark.asyncio +async def test_refresh_files_list(test_session, make_files_structure, test_directory): + make_files_structure(['foo.txt', 'bar.txt']) + + # Only foo.txt should have been refreshed. + await lib.refresh_files_list(['foo.txt']) + assert_files(test_session, ['foo.txt']) + + # Both files should be found + await lib.refresh_files_list(['bar.txt']) + assert_files(test_session, ['foo.txt', 'bar.txt']) + + # Both files should be unchanged. + await lib.refresh_files_list(['foo.txt', 'bar.txt']) + assert_files(test_session, ['foo.txt', 'bar.txt']) + + with pytest.raises(FileNotFoundError): + # Some files must be refreshed. + await lib.refresh_files_list([]) + + # Files that share the stem can also be refreshed. + (test_directory / 'foo.mp4').touch() + await lib.refresh_files_list(['foo.txt', 'bar.txt']) + assert_files(test_session, ['foo.txt', 'foo.mp4', 'bar.txt']) + + # Files that share the stem can also be ignored. + (test_directory / 'bar.mp4').touch() + await lib.refresh_files_list(['bar.txt'], include_files_near=False) + assert_files(test_session, ['foo.txt', 'foo.mp4', 'bar.txt']) + + # Deleted files are removed. + (test_directory / 'foo.mp4').unlink() + (test_directory / 'bar.txt').unlink() + await lib.refresh_files_list(['foo.txt', 'bar.txt'], include_files_near=False) + assert_files(test_session, ['foo.txt', ]) + # bar.mp4 is discovered near non-existent bar.txt. + await lib.refresh_files_list(['foo.txt', 'bar.txt']) + assert_files(test_session, ['foo.txt', 'bar.mp4']) + + @pytest.mark.asyncio async def test_refresh_cancel(test_session, make_files_structure, test_directory): """Refresh tasks can be canceled."""