From 7f094a562a834a2fe5d2a98150aac4d8ec7fda16 Mon Sep 17 00:00:00 2001 From: David Straub Date: Tue, 10 Dec 2024 22:31:21 +0100 Subject: [PATCH] More fixes --- gramps_webapi/api/html.py | 4 +- gramps_webapi/api/media_importer.py | 2 +- gramps_webapi/api/resources/bookmarks.py | 4 +- gramps_webapi/api/resources/delete.py | 2 +- gramps_webapi/api/resources/filters.py | 2 +- gramps_webapi/api/resources/util.py | 108 ++++++++++++----------- gramps_webapi/api/search/text.py | 2 + gramps_webapi/api/tasks.py | 6 +- gramps_webapi/api/util.py | 4 +- 9 files changed, 72 insertions(+), 62 deletions(-) diff --git a/gramps_webapi/api/html.py b/gramps_webapi/api/html.py index 7437a735..732d123c 100644 --- a/gramps_webapi/api/html.py +++ b/gramps_webapi/api/html.py @@ -23,8 +23,8 @@ from typing import Callable, Optional -import bleach -from bleach.css_sanitizer import CSSSanitizer +import bleach # type: ignore +from bleach.css_sanitizer import CSSSanitizer # type: ignore from gramps.gen.errors import HandleError from gramps.gen.lib import Note, NoteType, StyledText from gramps.plugins.lib.libhtml import Html diff --git a/gramps_webapi/api/media_importer.py b/gramps_webapi/api/media_importer.py index 7c5e0d5d..7f39a667 100644 --- a/gramps_webapi/api/media_importer.py +++ b/gramps_webapi/api/media_importer.py @@ -77,7 +77,7 @@ def _identify_missing_files(self) -> MissingFiles: obj for obj in self.objects if obj.handle not in handles_existing ] - missing_files = {} + missing_files: dict[str, list[dict[str, str]]] = {} for obj in objects_missing: if obj.checksum not in missing_files: missing_files[obj.checksum] = [] diff --git a/gramps_webapi/api/resources/bookmarks.py b/gramps_webapi/api/resources/bookmarks.py index 9cf51207..d04920a7 100644 --- a/gramps_webapi/api/resources/bookmarks.py +++ b/gramps_webapi/api/resources/bookmarks.py @@ -69,10 +69,10 @@ def _get_bookmarks_object(db_handle: DbReadBase, namespace: str) -> DbBookmarks: return result -def get_bookmarks(db_handle: DbReadBase, namespace: str) -> Optional[List]: +def get_bookmarks(db_handle: DbReadBase, namespace: str) -> list: """Return bookmarks for a namespace.""" bookmarks = _get_bookmarks_object(db_handle, namespace) - return bookmarks.get() + return bookmarks.get() or [] def create_bookmark(db_handle: DbReadBase, namespace: str, handle: str) -> None: diff --git a/gramps_webapi/api/resources/delete.py b/gramps_webapi/api/resources/delete.py index a3ede934..f73c6535 100644 --- a/gramps_webapi/api/resources/delete.py +++ b/gramps_webapi/api/resources/delete.py @@ -447,7 +447,7 @@ def delete_all_objects( db_handle: DbWriteBase, namespaces: Optional[List[str]] = None, progress_cb: Optional[Callable] = None, -) -> List[Dict[str, Any]]: +) -> None: """Delete all objects, optionally restricting to one or more types (namespaces).""" if progress_cb: total = get_total_number_of_objects(db_handle) diff --git a/gramps_webapi/api/resources/filters.py b/gramps_webapi/api/resources/filters.py index eab6ebad..1567abad 100644 --- a/gramps_webapi/api/resources/filters.py +++ b/gramps_webapi/api/resources/filters.py @@ -303,7 +303,7 @@ def delete(self, args: Dict, namespace: str, name: str) -> Response: custom_filters = filters.CustomFilters.get_filters(namespace) for custom_filter in custom_filters: if name == custom_filter.get_name(): - filter_set = set() + filter_set: set[GenericFilter] = set() self._find_dependent_filters(namespace, custom_filter, filter_set) if len(filter_set) > 1: if "force" not in args or not args["force"]: diff --git a/gramps_webapi/api/resources/util.py b/gramps_webapi/api/resources/util.py index 06db2d47..4feeceac 100644 --- a/gramps_webapi/api/resources/util.py +++ b/gramps_webapi/api/resources/util.py @@ -20,12 +20,13 @@ """Gramps utility functions.""" +from __future__ import annotations import json import os from hashlib import sha256 from http import HTTPStatus -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Any, cast, Optional, Union, Literal import gramps import jsonschema @@ -82,7 +83,7 @@ _ = glocale.translation.gettext -def get_person_by_handle(db_handle: DbReadBase, handle: Handle) -> Union[Person, Dict]: +def get_person_by_handle(db_handle: DbReadBase, handle: Handle) -> Union[Person, dict]: """Safe get person by handle.""" try: return db_handle.get_person_from_handle(handle) @@ -90,7 +91,7 @@ def get_person_by_handle(db_handle: DbReadBase, handle: Handle) -> Union[Person, return {} -def get_place_by_handle(db_handle: DbReadBase, handle: Handle) -> Union[Place, Dict]: +def get_place_by_handle(db_handle: DbReadBase, handle: Handle) -> Union[Place, dict]: """Safe get place by handle.""" try: return db_handle.get_place_from_handle(handle) @@ -99,8 +100,8 @@ def get_place_by_handle(db_handle: DbReadBase, handle: Handle) -> Union[Place, D def get_family_by_handle( - db_handle: DbReadBase, handle: Handle, args: Optional[Dict] = None -) -> Union[Family, Dict]: + db_handle: DbReadBase, handle: Handle, args: Optional[dict] = None +) -> Union[Family, dict]: """Get a family and optional extended attributes.""" try: obj = db_handle.get_family_from_handle(handle) @@ -117,7 +118,7 @@ def get_family_by_handle( def get_source_by_handle( - db_handle: DbReadBase, handle: Handle, args: Optional[Dict] = None + db_handle: DbReadBase, handle: Handle, args: Optional[dict] = None ) -> Source: """Get a source and optional extended attributes.""" args = args or {} @@ -140,9 +141,12 @@ def get_event_participants_for_handle( db_handle: DbReadBase, handle: Handle, locale: GrampsLocale = glocale, -) -> Dict[str, List[Tuple[EventRoleType, Union[Person, Family]]]]: +) -> dict[str, list[tuple[EventRoleType, Union[Person, Family]]]]: """Get event participants given a handle.""" - result = {"people": [], "families": []} + result: dict[Literal["people", "families"], list[tuple[Any, Person | Family]]] = { + "people": [], + "families": [], + } seen = set() # to avoid duplicates for class_name, backref_handle in db_handle.find_backlink_handles( handle, include_classes=["Person", "Family"] @@ -181,7 +185,7 @@ def get_event_participants_profile_for_handle( db_handle: DbReadBase, handle: Handle, locale: GrampsLocale = glocale, -) -> Dict: +) -> dict: """Get event participants given a handle.""" event_participants = get_event_participants_for_handle( db_handle=db_handle, @@ -192,13 +196,13 @@ def get_event_participants_profile_for_handle( for role, person in event_participants["people"]: person_profile = get_person_profile_for_object( - db_handle, person, args=[], locale=locale + db_handle, cast(Person, person), args=[], locale=locale ) role_str = locale.translation.sgettext(role.xml_str()) result["people"].append({"role": role_str, "person": person_profile}) - for role, person in event_participants["families"]: + for role, family in event_participants["families"]: person_profile = get_family_profile_for_object( - db_handle, person, args=[], locale=locale + db_handle, cast(Family, family), args=[], locale=locale ) role_str = locale.translation.sgettext(role.xml_str()) result["families"].append({"role": role_str, "family": person_profile}) @@ -220,12 +224,12 @@ def get_event_summary_from_object( def get_event_profile_for_object( db_handle: DbReadBase, event: Event, - args: List, + args: list, base_event: Union[Event, None] = None, label: str = "span", locale: GrampsLocale = glocale, role: Optional[str] = None, -) -> Dict: +) -> dict: """Get event profile given an Event.""" result = { "type": locale.translation.sgettext(event.type.xml_str()), @@ -255,12 +259,12 @@ def get_event_profile_for_object( def get_event_profile_for_handle( db_handle: DbReadBase, handle: Handle, - args: List, + args: list, base_event: Union[Event, None] = None, label: str = "span", locale: GrampsLocale = glocale, role: Optional[str] = None, -) -> Dict: +) -> dict: """Get event profile given a handle.""" try: obj = db_handle.get_event_from_handle(handle) @@ -280,9 +284,9 @@ def get_event_profile_for_handle( def get_birth_profile( db_handle: DbReadBase, person: Person, - args: Union[List, None] = None, + args: Union[list, None] = None, locale: GrampsLocale = glocale, -) -> Tuple[Dict, Union[Event, None]]: +) -> tuple[dict, Union[Event, None]]: """Return best available birth information for a person.""" event = get_birth_or_fallback(db_handle, person) if event is None: @@ -297,9 +301,9 @@ def get_birth_profile( def get_death_profile( db_handle: DbReadBase, person: Person, - args: Union[List, None] = None, + args: Union[list, None] = None, locale: GrampsLocale = glocale, -) -> Tuple[Dict, Union[Event, None]]: +) -> tuple[dict, Union[Event, None]]: """Return best available death information for a person.""" event = get_death_or_fallback(db_handle, person) if event is None: @@ -314,9 +318,9 @@ def get_death_profile( def get_marriage_profile( db_handle: DbReadBase, family: Family, - args: Union[List, None] = None, + args: Union[list, None] = None, locale: GrampsLocale = glocale, -) -> Tuple[Dict, Union[Event, None]]: +) -> tuple[dict, Union[Event, None]]: """Return best available marriage information for a couple.""" event = get_marriage_or_fallback(db_handle, family) if event is None: @@ -331,9 +335,9 @@ def get_marriage_profile( def get_divorce_profile( db_handle: DbReadBase, family: Family, - args: Union[List, None] = None, + args: Union[list, None] = None, locale: GrampsLocale = glocale, -) -> Tuple[Dict, Union[Event, None]]: +) -> tuple[dict, Union[Event, None]]: """Return best available divorce information for a couple.""" event = get_divorce_or_fallback(db_handle, family) if event is None: @@ -347,7 +351,7 @@ def get_divorce_profile( def _format_place_type( place_type: PlaceType, locale: GrampsLocale = glocale -) -> Dict[str, Any]: +) -> dict[str, Any]: """Format a place type.""" return locale.translation.sgettext(place_type.xml_str()) @@ -357,7 +361,7 @@ def get_place_profile_for_object( place: Place, locale: GrampsLocale = glocale, parent_places: bool = True, -) -> Dict[str, Any]: +) -> dict[str, Any]: """Get place profile given a Place.""" latitude, longitude = conv_lat_lon(place.lat, place.long, format="D.D8") profile = { @@ -401,7 +405,7 @@ def get_place_profile_for_handle( handle: Handle, locale: GrampsLocale = glocale, parent_places: bool = True, -) -> Union[Media, Dict]: +) -> Union[Media, dict]: """Get place profile given a handle.""" obj = get_place_by_handle(db_handle, handle) return get_place_profile_for_object( @@ -410,7 +414,7 @@ def get_place_profile_for_handle( def get_person_profile_for_object( - db_handle: DbReadBase, person: Person, args: List, locale: GrampsLocale = glocale + db_handle: DbReadBase, person: Person, args: list, locale: GrampsLocale = glocale ) -> Person: """Get person profile given a Person.""" options = [] @@ -484,8 +488,8 @@ def get_person_profile_for_object( def get_person_profile_for_handle( - db_handle: DbReadBase, handle: Handle, args: List, locale: GrampsLocale = glocale -) -> Union[Person, Dict]: + db_handle: DbReadBase, handle: Handle, args: list, locale: GrampsLocale = glocale +) -> Union[Person, dict]: """Get person profile given a handle.""" try: obj = db_handle.get_person_from_handle(handle) @@ -495,7 +499,7 @@ def get_person_profile_for_handle( def get_family_profile_for_object( - db_handle: DbReadBase, family: Family, args: List, locale: GrampsLocale = glocale + db_handle: DbReadBase, family: Family, args: list, locale: GrampsLocale = glocale ) -> Family: """Get family profile given a Family.""" options = [] @@ -566,8 +570,8 @@ def get_family_profile_for_object( def get_family_profile_for_handle( - db_handle: DbReadBase, handle: Handle, args: List, locale: GrampsLocale = glocale -) -> Union[Family, Dict]: + db_handle: DbReadBase, handle: Handle, args: list, locale: GrampsLocale = glocale +) -> Union[Family, dict]: """Get family profile given a handle.""" try: obj = db_handle.get_family_from_handle(handle) @@ -579,7 +583,7 @@ def get_family_profile_for_handle( def get_citation_profile_for_object( db_handle: DbReadBase, citation: Citation, - args: List, + args: list, locale: GrampsLocale = glocale, ) -> Citation: """Get citation profile given a Citation.""" @@ -598,8 +602,8 @@ def get_citation_profile_for_object( def get_citation_profile_for_handle( - db_handle: DbReadBase, handle: Handle, args: List, locale: GrampsLocale = glocale -) -> Union[Family, Dict]: + db_handle: DbReadBase, handle: Handle, args: list, locale: GrampsLocale = glocale +) -> Union[Family, dict]: """Get citation profile given a handle.""" try: obj = db_handle.get_citation_from_handle(handle) @@ -609,7 +613,7 @@ def get_citation_profile_for_handle( def get_media_profile_for_object( - db_handle: DbReadBase, media: Media, args: List, locale: GrampsLocale = glocale + db_handle: DbReadBase, media: Media, args: list, locale: GrampsLocale = glocale ) -> Media: """Get media profile given Media.""" return { @@ -619,8 +623,8 @@ def get_media_profile_for_object( def get_media_profile_for_handle( - db_handle: DbReadBase, handle: Handle, args: List, locale: GrampsLocale = glocale -) -> Union[Media, Dict]: + db_handle: DbReadBase, handle: Handle, args: list, locale: GrampsLocale = glocale +) -> Union[Media, dict]: """Get media profile given a handle.""" try: obj = db_handle.get_media_from_handle(handle) @@ -638,8 +642,8 @@ def catch_handle_error(method, handle): def get_extended_attributes( - db_handle: DbReadBase, obj: GrampsObject, args: Optional[Dict] = None -) -> Dict: + db_handle: DbReadBase, obj: GrampsObject, args: Optional[dict] = None +) -> dict: """Get extended attributes for a GrampsObject.""" args = args or {} result = {} @@ -704,7 +708,7 @@ def get_extended_attributes( return result -def get_backlinks(db_handle: DbReadBase, handle: Handle) -> Dict[str, List[Handle]]: +def get_backlinks(db_handle: DbReadBase, handle: Handle) -> dict[str, list[Handle]]: """Get backlinks to a handle. Will return a dictionary of the form @@ -737,7 +741,7 @@ def get_reference_profile_for_object( db_handle: DbReadBase, obj: GrampsObject, locale: GrampsLocale = glocale, -) -> Dict: +) -> dict: """Return reference profiles for an object.""" profile = {} # get backlink handles @@ -779,7 +783,7 @@ def get_reference_profile_for_object( return profile -def get_rating(db_handle: DbReadBase, obj: GrampsObject) -> Tuple[int, int]: +def get_rating(db_handle: DbReadBase, obj: GrampsObject) -> tuple[int, int]: """Return rating based on citations.""" count = 0 confidence = 0 @@ -876,7 +880,7 @@ def add_family_update_refs( db_handle.commit_person(child, trans) -def validate_object_dict(obj_dict: Dict[str, Any]) -> bool: +def validate_object_dict(obj_dict: dict[str, Any]) -> bool: """Validate a dict representation of a Gramps object vs. its schema.""" try: obj_cls = getattr(gramps.gen.lib, obj_dict["_class"]) @@ -900,7 +904,7 @@ def xml_to_locale(gramps_type_name: str, string: str) -> str: return str(typ) -def fix_object_dict(object_dict: Dict, class_name: Optional[str] = None): +def fix_object_dict(object_dict: dict, class_name: Optional[str] = None): """Restore a Gramps object in simplified representation to its full form. This restores in particular: @@ -1149,7 +1153,7 @@ def hash_object(obj: GrampsObject) -> str: return sha256(data).hexdigest() -def filter_missing_files(objects: List[Media]) -> List[Media]: +def filter_missing_files(objects: list[Media]) -> list[Media]: """Filter media objects returning only ones where the file is missing.""" tree = get_tree_from_jwt() db_handle = get_db_handle() @@ -1160,8 +1164,8 @@ def filter_missing_files(objects: List[Media]) -> List[Media]: def get_missing_media_file_handles( - db_handle: DbReadBase, handles: List[str] -) -> List[str]: + db_handle: DbReadBase, handles: list[str] +) -> list[str]: """Filter media handles returning only ones where the file is missing.""" objects = [db_handle.get_media_from_handle(handle) for handle in handles] objects_missing = filter_missing_files(objects) @@ -1174,7 +1178,7 @@ def get_one_relationship( person2: Person, depth: int, locale: GrampsLocale = glocale, -) -> Tuple[str, int, int]: +) -> tuple[str, int, int]: """Get a relationship string and the number of generations between the people.""" calc = get_relationship_calculator(reinit=True, clocale=locale) # the relationship calculation can be slow when depth is set to a large value @@ -1193,7 +1197,7 @@ def get_one_relationship( ) -def get_importers(extension: str = None): +def get_importers(extension: str | None = None): """Extract and return list of importers.""" importers = [] plugin_manager = BasePluginManager.get_instance() @@ -1238,7 +1242,7 @@ def run_import( def dry_run_import( file_name: FilenameOrPath, -) -> Optional[Dict[str, int]]: +) -> Optional[dict[str, int]]: """Import a file into an in-memory database and returns object counts.""" db_handle: DbReadBase = import_as_dict(filename=file_name, user=User()) if db_handle is None: diff --git a/gramps_webapi/api/search/text.py b/gramps_webapi/api/search/text.py index f582580e..038cec80 100644 --- a/gramps_webapi/api/search/text.py +++ b/gramps_webapi/api/search/text.py @@ -158,6 +158,7 @@ def obj_strings_from_handle( ) -> Optional[Dict[str, Any]]: """Return object strings from a handle and Gramps class name.""" query_method = db_handle.method("get_%s_from_handle", class_name) + assert query_method is not None # type checker obj = query_method(handle) return obj_strings_from_object( db_handle=db_handle, class_name=class_name, obj=obj, semantic=semantic @@ -192,6 +193,7 @@ def iter_obj_strings( for class_name in PRIMARY_GRAMPS_OBJECTS: plural_name = GRAMPS_OBJECT_PLURAL[class_name] iter_method = db_handle.method("iter_%s", plural_name) + assert iter_method is not None # type checker for obj in iter_method(): obj_strings = obj_strings_from_object( db_handle, class_name, obj, semantic=semantic diff --git a/gramps_webapi/api/tasks.py b/gramps_webapi/api/tasks.py index 9fa426ef..ecd9b2db 100644 --- a/gramps_webapi/api/tasks.py +++ b/gramps_webapi/api/tasks.py @@ -30,6 +30,8 @@ from celery.result import AsyncResult from flask import current_app +from gramps_webapi.api.search.indexer import SearchIndexer, SemanticSearchIndexer + from ..auth import get_owner_emails from .check import check_database from .emails import email_confirm_email, email_new_user, email_reset_pw @@ -111,7 +113,9 @@ def _search_reindex_full( ) -> None: """Rebuild the search index.""" if semantic: - indexer = get_semantic_search_indexer(tree) + indexer: SearchIndexer | SemanticSearchIndexer = get_semantic_search_indexer( + tree + ) else: indexer = get_search_indexer(tree) db = get_db_outside_request( diff --git a/gramps_webapi/api/util.py b/gramps_webapi/api/util.py index c86b5b07..410c056b 100644 --- a/gramps_webapi/api/util.py +++ b/gramps_webapi/api/util.py @@ -31,7 +31,7 @@ from email.message import EmailMessage from email.utils import make_msgid from http import HTTPStatus -from typing import Any, BinaryIO, Never, Sequence +from typing import Any, BinaryIO, NoReturn, Sequence from celery import Task from flask import ( @@ -704,7 +704,7 @@ def check_quota_ai(requested: int, tree: str | None = None) -> None: abort_with_message(405, "Not allowed by AI quota") -def abort_with_message(status: int, message: str) -> Never: +def abort_with_message(status: int, message: str) -> NoReturn: """Abort with a JSON response.""" payload = {"error": {"code": status, "message": message}} response = Response(