diff --git a/pyproject.toml b/pyproject.toml index 1a8deefd78..f5141faed8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -75,11 +75,6 @@ disallow_incomplete_defs = false disallow_untyped_decorators = false disallow_untyped_defs = false module = [ - "palace.manager.api.admin.admin_authentication_provider", - "palace.manager.api.admin.model.custom_lists", - "palace.manager.api.admin.password_admin_authentication_provider", - "palace.manager.api.admin.routes", - "palace.manager.api.admin.validator", "palace.manager.api.annotations", "palace.manager.api.app", "palace.manager.api.authentication.base", diff --git a/src/palace/manager/api/admin/admin_authentication_provider.py b/src/palace/manager/api/admin/admin_authentication_provider.py index 02103857b9..3127e9af0b 100644 --- a/src/palace/manager/api/admin/admin_authentication_provider.py +++ b/src/palace/manager/api/admin/admin_authentication_provider.py @@ -1,9 +1,17 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from palace.manager.sqlalchemy.model.admin import Admin + + class AdminAuthenticationProvider: - def sign_in_template(self, redirect_url): + def sign_in_template(self, redirect_url: str | None) -> str: # Returns HTML to be rendered on the sign in page for # this authentication provider. raise NotImplementedError() - def active_credentials(self, admin): + def active_credentials(self, admin: Admin) -> bool: # Returns True if the admin's credentials are not expired. raise NotImplementedError() diff --git a/src/palace/manager/api/admin/controller/catalog_services.py b/src/palace/manager/api/admin/controller/catalog_services.py index 3f6abfff8d..2d09feb7c5 100644 --- a/src/palace/manager/api/admin/controller/catalog_services.py +++ b/src/palace/manager/api/admin/controller/catalog_services.py @@ -7,7 +7,10 @@ UpdatedLibrarySettingsTuple, ) from palace.manager.api.admin.form_data import ProcessFormData -from palace.manager.api.admin.problem_details import MULTIPLE_SERVICES_FOR_LIBRARY +from palace.manager.api.admin.problem_details import ( + MISSING_SERVICE, + MULTIPLE_SERVICES_FOR_LIBRARY, +) from palace.manager.integration.catalog.marc.exporter import MarcExporter from palace.manager.integration.goals import Goals from palace.manager.integration.settings import BaseSettings @@ -102,6 +105,10 @@ def process_post(self) -> Response | ProblemDetail: return Response(str(catalog_service.id), response_code) - def process_delete(self, service_id: int) -> Response: + def process_delete(self, service_id: int | str) -> Response | ProblemDetail: self.require_system_admin() - return self.delete_service(service_id) + try: + sid = int(service_id) if isinstance(service_id, str) else service_id + except ValueError: + return MISSING_SERVICE + return self.delete_service(sid) diff --git a/src/palace/manager/api/admin/controller/collection_settings.py b/src/palace/manager/api/admin/controller/collection_settings.py index 85c6e8dae2..040a4a09c6 100644 --- a/src/palace/manager/api/admin/controller/collection_settings.py +++ b/src/palace/manager/api/admin/controller/collection_settings.py @@ -13,6 +13,7 @@ from palace.manager.api.admin.problem_details import ( CANNOT_DELETE_COLLECTION_WITH_CHILDREN, MISSING_COLLECTION, + MISSING_IDENTIFIER, MISSING_PARENT, MISSING_SERVICE, PROTOCOL_DOES_NOT_SUPPORT_PARENTS, @@ -171,13 +172,17 @@ def process_deleted_libraries( reap_unassociated_loans.delay() reap_unassociated_holds.delay() - def process_delete(self, service_id: int) -> Response | ProblemDetail: + def process_delete(self, service_id: int | str) -> Response | ProblemDetail: self.require_system_admin() + try: + sid = int(service_id) if isinstance(service_id, str) else service_id + except ValueError: + return MISSING_SERVICE integration = get_one( self._db, IntegrationConfiguration, - id=service_id, + id=sid, goal=self.registry.goal, ) if not integration: @@ -196,8 +201,13 @@ def process_delete(self, service_id: int) -> Response | ProblemDetail: return Response("Deleted", 200) def process_collection_self_tests( - self, identifier: int | None + self, identifier: int | str | None ) -> Response | ProblemDetail: + if identifier is not None and isinstance(identifier, str): + try: + identifier = int(identifier) + except ValueError: + return MISSING_IDENTIFIER return self.process_self_tests(identifier) def run_self_tests( diff --git a/src/palace/manager/api/admin/controller/custom_lists.py b/src/palace/manager/api/admin/controller/custom_lists.py index 260d913080..a11b99259e 100644 --- a/src/palace/manager/api/admin/controller/custom_lists.py +++ b/src/palace/manager/api/admin/controller/custom_lists.py @@ -313,8 +313,12 @@ def url_fn(after: int) -> str: return url_fn def custom_list( - self, list_id: int + self, list_id: int | str ) -> Response | dict[str, Any] | ProblemDetail | None: + try: + list_id = int(list_id) if isinstance(list_id, str) else list_id + except ValueError: + return MISSING_CUSTOM_LIST library = get_request_library() self.require_librarian(library) data_source = DataSource.lookup(self._db, DataSource.LIBRARY_STAFF) @@ -398,9 +402,15 @@ def custom_list( return None def share_locally( - self, customlist_id: int + self, customlist_id: int | str ) -> ProblemDetail | dict[str, int] | Response: """Share this customlist with all libraries on this local CM""" + try: + customlist_id = ( + int(customlist_id) if isinstance(customlist_id, str) else customlist_id + ) + except ValueError: + return MISSING_CUSTOM_LIST if not customlist_id: return INVALID_INPUT customlist = get_one(self._db, CustomList, id=customlist_id) diff --git a/src/palace/manager/api/admin/controller/dashboard.py b/src/palace/manager/api/admin/controller/dashboard.py index 367767cc16..7541e35600 100644 --- a/src/palace/manager/api/admin/controller/dashboard.py +++ b/src/palace/manager/api/admin/controller/dashboard.py @@ -14,6 +14,7 @@ from palace.manager.api.local_analytics_exporter import LocalAnalyticsExporter from palace.manager.api.util.flask import get_request_library from palace.manager.sqlalchemy.model.admin import Admin +from palace.manager.util.problem_detail import ProblemDetail class DashboardController(CirculationManagerController): @@ -25,7 +26,7 @@ def stats( def bulk_circulation_events( self, analytics_exporter: LocalAnalyticsExporter | None = None - ) -> tuple[str, str, str, str | None]: + ) -> tuple[str | ProblemDetail, str, str, str | None]: date_format = "%Y-%m-%d" def get_date(field: str) -> date: diff --git a/src/palace/manager/api/admin/controller/discovery_services.py b/src/palace/manager/api/admin/controller/discovery_services.py index c8fb4c1b31..504270ee62 100644 --- a/src/palace/manager/api/admin/controller/discovery_services.py +++ b/src/palace/manager/api/admin/controller/discovery_services.py @@ -7,7 +7,10 @@ IntegrationSettingsController, ) from palace.manager.api.admin.form_data import ProcessFormData -from palace.manager.api.admin.problem_details import INTEGRATION_URL_ALREADY_IN_USE +from palace.manager.api.admin.problem_details import ( + INTEGRATION_URL_ALREADY_IN_USE, + MISSING_SERVICE, +) from palace.manager.integration.discovery.opds_registration import ( OpdsRegistrationService, ) @@ -78,10 +81,14 @@ def process_post(self) -> Response | ProblemDetail: return Response(str(service.id), response_code) - def process_delete(self, service_id: int) -> Response | ProblemDetail: + def process_delete(self, service_id: int | str) -> Response | ProblemDetail: self.require_system_admin() try: - return self.delete_service(service_id) + sid = int(service_id) if isinstance(service_id, str) else service_id + except ValueError: + return MISSING_SERVICE + try: + return self.delete_service(sid) except ProblemDetailException as e: self._db.rollback() return e.problem_detail diff --git a/src/palace/manager/api/admin/controller/lanes.py b/src/palace/manager/api/admin/controller/lanes.py index 9dc909a387..7e29532eb6 100644 --- a/src/palace/manager/api/admin/controller/lanes.py +++ b/src/palace/manager/api/admin/controller/lanes.py @@ -165,7 +165,15 @@ def lanes_for_parent(parent: Lane | None) -> list[dict[str, Any]]: return Response(str(lane.id), 200) raise RuntimeError("Unsupported method") - def lane(self, lane_identifier: int) -> Response | ProblemDetail: + def lane(self, lane_identifier: int | str) -> Response | ProblemDetail: + try: + lane_identifier = ( + int(lane_identifier) + if isinstance(lane_identifier, str) + else lane_identifier + ) + except ValueError: + return MISSING_LANE if flask.request.method == "DELETE": library = get_request_library() self.require_library_manager(library) @@ -197,7 +205,15 @@ def _check_lane_name_unique( LANE_WITH_PARENT_AND_DISPLAY_NAME_ALREADY_EXISTS ) - def show_lane(self, lane_identifier: int) -> Response | ProblemDetail: + def show_lane(self, lane_identifier: int | str) -> Response | ProblemDetail: + try: + lane_identifier = ( + int(lane_identifier) + if isinstance(lane_identifier, str) + else lane_identifier + ) + except ValueError: + return MISSING_LANE library = get_request_library() self.require_library_manager(library) @@ -209,7 +225,15 @@ def show_lane(self, lane_identifier: int) -> Response | ProblemDetail: lane.visible = True return Response(str(_("Success")), 200) - def hide_lane(self, lane_identifier: int) -> Response | ProblemDetail: + def hide_lane(self, lane_identifier: int | str) -> Response | ProblemDetail: + try: + lane_identifier = ( + int(lane_identifier) + if isinstance(lane_identifier, str) + else lane_identifier + ) + except ValueError: + return MISSING_LANE library = get_request_library() self.require_library_manager(library) diff --git a/src/palace/manager/api/admin/controller/metadata_services.py b/src/palace/manager/api/admin/controller/metadata_services.py index 0ed40cbf1c..0369a3b0ca 100644 --- a/src/palace/manager/api/admin/controller/metadata_services.py +++ b/src/palace/manager/api/admin/controller/metadata_services.py @@ -8,7 +8,11 @@ IntegrationSettingsSelfTestsController, ) from palace.manager.api.admin.form_data import ProcessFormData -from palace.manager.api.admin.problem_details import DUPLICATE_INTEGRATION +from palace.manager.api.admin.problem_details import ( + DUPLICATE_INTEGRATION, + MISSING_IDENTIFIER, + MISSING_SERVICE, +) from palace.manager.core.selftest import HasSelfTests from palace.manager.integration.base import HasLibraryIntegrationConfiguration from palace.manager.integration.metadata.base import MetadataServiceType @@ -85,9 +89,13 @@ def process_post(self) -> Response | ProblemDetail: return Response(str(metadata_service.id), response_code) - def process_delete(self, service_id: int) -> Response: + def process_delete(self, service_id: int | str) -> Response | ProblemDetail: self.require_system_admin() - return self.delete_service(service_id) + try: + sid = int(service_id) if isinstance(service_id, str) else service_id + except ValueError: + return MISSING_SERVICE + return self.delete_service(sid) def run_self_tests( self, integration: IntegrationConfiguration @@ -103,6 +111,11 @@ def run_self_tests( return None def process_metadata_service_self_tests( - self, identifier: int | None + self, identifier: int | str | None ) -> Response | ProblemDetail: + if identifier is not None and isinstance(identifier, str): + try: + identifier = int(identifier) + except ValueError: + return MISSING_IDENTIFIER return self.process_self_tests(identifier) diff --git a/src/palace/manager/api/admin/controller/patron_auth_services.py b/src/palace/manager/api/admin/controller/patron_auth_services.py index a652f591d3..9d346cc697 100644 --- a/src/palace/manager/api/admin/controller/patron_auth_services.py +++ b/src/palace/manager/api/admin/controller/patron_auth_services.py @@ -13,6 +13,8 @@ from palace.manager.api.admin.form_data import ProcessFormData from palace.manager.api.admin.problem_details import ( FAILED_TO_RUN_SELF_TESTS, + MISSING_IDENTIFIER, + MISSING_SERVICE, MULTIPLE_BASIC_AUTH_SERVICES, ) from palace.manager.api.authentication.base import AuthenticationProviderType @@ -120,17 +122,26 @@ def process_updated_libraries( for integration, _ in libraries: self.library_integration_validation(integration) - def process_delete(self, service_id: int) -> Response | ProblemDetail: + def process_delete(self, service_id: int | str) -> Response | ProblemDetail: self.require_system_admin() try: - return self.delete_service(service_id) + sid = int(service_id) if isinstance(service_id, str) else service_id + except ValueError: + return MISSING_SERVICE + try: + return self.delete_service(sid) except ProblemDetailException as e: self._db.rollback() return e.problem_detail def process_patron_auth_service_self_tests( - self, identifier: int | None + self, identifier: int | str | None ) -> Response | ProblemDetail: + if identifier is not None and isinstance(identifier, str): + try: + identifier = int(identifier) + except ValueError: + return MISSING_IDENTIFIER return self.process_self_tests(identifier) def get_prior_test_results( diff --git a/src/palace/manager/api/admin/model/custom_lists.py b/src/palace/manager/api/admin/model/custom_lists.py index a96230e2a5..49361bd1f9 100644 --- a/src/palace/manager/api/admin/model/custom_lists.py +++ b/src/palace/manager/api/admin/model/custom_lists.py @@ -1,3 +1,5 @@ +from typing import Any + from pydantic import NonNegativeInt from palace.manager.util.flask_util import CustomBaseModel @@ -11,10 +13,10 @@ class CustomListSharePostResponse(CustomBaseModel): class CustomListPostRequest(CustomBaseModel): name: str id: NonNegativeInt | None = None - entries: list[dict] = [] + entries: list[dict[str, Any]] = [] collections: list[int] = [] - deletedEntries: list[dict] = [] + deletedEntries: list[dict[str, Any]] = [] # For auto updating lists auto_update: bool = False - auto_update_query: dict | None = None - auto_update_facets: dict | None = None + auto_update_query: dict[str, Any] | None = None + auto_update_facets: dict[str, Any] | None = None diff --git a/src/palace/manager/api/admin/password_admin_authentication_provider.py b/src/palace/manager/api/admin/password_admin_authentication_provider.py index 058ef0da31..e93f76cfaa 100644 --- a/src/palace/manager/api/admin/password_admin_authentication_provider.py +++ b/src/palace/manager/api/admin/password_admin_authentication_provider.py @@ -1,5 +1,7 @@ -from flask import render_template, url_for +from flask import Response, render_template, url_for from sqlalchemy.orm.session import Session +from werkzeug.datastructures import ImmutableMultiDict +from werkzeug.wrappers.response import Response as WerkzeugResponse from palace.manager.api.admin.admin_authentication_provider import ( AdminAuthenticationProvider, @@ -28,7 +30,7 @@ def __init__(self, send_email: SendEmailCallable): def get_secret_key(db: Session) -> str: return Key.get_key(db, KeyType.ADMIN_SECRET_KEY, raise_exception=True).value - def sign_in_template(self, redirect): + def sign_in_template(self, redirect: str | None) -> str: password_sign_in_url = url_for("password_auth") forgot_password_url = url_for("admin_forgot_password") return render_template( @@ -44,7 +46,9 @@ def sign_in_template(self, redirect): ) @staticmethod - def forgot_password_template(redirect): + def forgot_password_template( + redirect: str | None | Response | WerkzeugResponse, + ) -> str: forgot_password_url = url_for("admin_forgot_password") return render_template( "admin/auth/forgot-password.html.jinja2", @@ -56,7 +60,11 @@ def forgot_password_template(redirect): ) @staticmethod - def reset_password_template(reset_password_token, admin_id, redirect): + def reset_password_template( + reset_password_token: str, + admin_id: int, + redirect: str | None | Response | WerkzeugResponse, + ) -> str: reset_password_url = url_for( "admin_reset_password", reset_password_token=reset_password_token, @@ -71,7 +79,13 @@ def reset_password_template(reset_password_token, admin_id, redirect): button_style=button_style, ) - def sign_in(self, _db, request={}): + def sign_in( + self, + _db: Session, + request: ImmutableMultiDict[str, str] | dict[str, str | None] | None = None, + ) -> tuple[dict[str, str], str | None] | tuple[ProblemDetail, None]: + if request is None: + request = {} email = request.get("email") password = request.get("password") redirect_url = request.get("redirect") @@ -91,7 +105,7 @@ def sign_in(self, _db, request={}): return INVALID_ADMIN_CREDENTIALS, None - def active_credentials(self, admin): + def active_credentials(self, admin: Admin) -> bool: # Admins who have a password are always active. return True diff --git a/src/palace/manager/api/admin/routes.py b/src/palace/manager/api/admin/routes.py index 817fa71e65..46216499a9 100644 --- a/src/palace/manager/api/admin/routes.py +++ b/src/palace/manager/api/admin/routes.py @@ -1,7 +1,9 @@ +# Decorators from palace.manager.api.routes and core.app_server are untyped. +# mypy: disallow_untyped_decorators=false from collections.abc import Callable from datetime import timedelta from functools import wraps -from typing import ParamSpec, TypeVar +from typing import Any, ParamSpec, TypeVar import flask from flask import Response, make_response, redirect, request, url_for @@ -31,9 +33,11 @@ T = TypeVar("T") -def allows_admin_auth_setup(f): +def allows_admin_auth_setup( + f: Callable[..., Any], +) -> Callable[..., Any]: @wraps(f) - def decorated(*args, **kwargs): + def decorated(*args: Any, **kwargs: Any) -> Any: setting_up = app.manager.admin_sign_in_controller.admin_auth_providers == [] return f(*args, setting_up=setting_up, **kwargs) @@ -61,9 +65,9 @@ def wrapper(*args: P.args, **kwargs: P.kwargs) -> T | ProblemDetail: return wrapper -def requires_admin(f): +def requires_admin(f: Callable[..., Any]) -> Callable[..., Any]: @wraps(f) - def decorated(*args, **kwargs): + def decorated(*args: Any, **kwargs: Any) -> Any: if "setting_up" in kwargs: # If the function also requires a CSRF token, # setting_up needs to stay in the arguments for @@ -89,11 +93,11 @@ def decorated(*args, **kwargs): return decorated -def requires_csrf_token(f): +def requires_csrf_token(f: Callable[..., Any]) -> Callable[..., Any]: f.__dict__["requires_csrf_token"] = True @wraps(f) - def decorated(*args, **kwargs): + def decorated(*args: Any, **kwargs: Any) -> Any: if "setting_up" in kwargs: setting_up = kwargs.pop("setting_up") else: @@ -107,9 +111,11 @@ def decorated(*args, **kwargs): return decorated -def returns_json_or_response_or_problem_detail(f): +def returns_json_or_response_or_problem_detail( + f: Callable[..., Any], +) -> Callable[..., Any]: @wraps(f) - def decorated(*args, **kwargs): + def decorated(*args: Any, **kwargs: Any) -> Any: try: v = f(*args, **kwargs) except BaseProblemDetailException as ex: @@ -126,33 +132,33 @@ def decorated(*args, **kwargs): @app.route("/admin/sign_in_with_password", methods=["POST"]) @returns_problem_detail -def password_auth(): +def password_auth() -> Any: return app.manager.admin_sign_in_controller.password_sign_in() @app.route("/admin/sign_in") @returns_problem_detail -def admin_sign_in(): +def admin_sign_in() -> Any: return app.manager.admin_sign_in_controller.sign_in() @app.route("/admin/sign_out") @returns_problem_detail @requires_admin -def admin_sign_out(): +def admin_sign_out() -> Any: return app.manager.admin_sign_in_controller.sign_out() @app.route("/admin/change_password", methods=["POST"]) @returns_problem_detail @requires_admin -def admin_change_password(): +def admin_change_password() -> Any: return app.manager.admin_sign_in_controller.change_password() @app.route("/admin/forgot_password", methods=["GET", "POST"]) @returns_problem_detail -def admin_forgot_password(): +def admin_forgot_password() -> Any: return app.manager.admin_reset_password_controller.forgot_password() @@ -160,9 +166,9 @@ def admin_forgot_password(): "/admin/reset_password//", methods=["GET", "POST"] ) @returns_problem_detail -def admin_reset_password(reset_password_token, admin_id): +def admin_reset_password(reset_password_token: str, admin_id: str) -> Any: return app.manager.admin_reset_password_controller.reset_password( - reset_password_token, admin_id + reset_password_token, int(admin_id) if admin_id.isdigit() else 0 ) @@ -170,7 +176,7 @@ def admin_reset_password(reset_password_token, admin_id): @has_library @returns_problem_detail @requires_admin -def work_details(identifier_type, identifier): +def work_details(identifier_type: str, identifier: str) -> Any: return app.manager.admin_work_controller.details(identifier_type, identifier) @@ -180,7 +186,7 @@ def work_details(identifier_type, identifier): @has_library @returns_json_or_response_or_problem_detail @requires_admin -def work_classifications(identifier_type, identifier): +def work_classifications(identifier_type: str, identifier: str) -> Any: return app.manager.admin_work_controller.classifications( identifier_type, identifier ) @@ -193,7 +199,7 @@ def work_classifications(identifier_type, identifier): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def work_custom_lists(identifier_type, identifier): +def work_custom_lists(identifier_type: str, identifier: str) -> Any: return app.manager.admin_work_controller.custom_lists(identifier_type, identifier) @@ -204,7 +210,7 @@ def work_custom_lists(identifier_type, identifier): @returns_problem_detail @requires_admin @requires_csrf_token -def edit(identifier_type, identifier): +def edit(identifier_type: str, identifier: str) -> Any: return app.manager.admin_work_controller.edit(identifier_type, identifier) @@ -215,7 +221,7 @@ def edit(identifier_type, identifier): @returns_problem_detail @requires_admin @requires_csrf_token -def suppress_for_library(identifier_type, identifier): +def suppress_for_library(identifier_type: str, identifier: str) -> Any: return app.manager.admin_work_controller.suppress(identifier_type, identifier) @@ -226,7 +232,7 @@ def suppress_for_library(identifier_type, identifier): @returns_problem_detail @requires_admin @requires_csrf_token -def unsuppress_for_library(identifier_type, identifier): +def unsuppress_for_library(identifier_type: str, identifier: str) -> Any: return app.manager.admin_work_controller.unsuppress(identifier_type, identifier) @@ -238,7 +244,7 @@ def unsuppress_for_library(identifier_type, identifier): @returns_problem_detail @requires_admin @requires_csrf_token -def suppress_deprecated(identifier_type, identifier): +def suppress_deprecated(identifier_type: str, identifier: str) -> Any: return app.manager.admin_work_controller.suppress(identifier_type, identifier) @@ -250,7 +256,7 @@ def suppress_deprecated(identifier_type, identifier): @returns_problem_detail @requires_admin @requires_csrf_token -def unsuppress_deprecated(identifier_type, identifier): +def unsuppress_deprecated(identifier_type: str, identifier: str) -> Any: return app.manager.admin_work_controller.unsuppress(identifier_type, identifier) @@ -259,7 +265,7 @@ def unsuppress_deprecated(identifier_type, identifier): @returns_problem_detail @requires_admin @requires_csrf_token -def refresh(identifier_type, identifier): +def refresh(identifier_type: str, identifier: str) -> Any: return app.manager.admin_work_controller.refresh_metadata( identifier_type, identifier ) @@ -273,7 +279,7 @@ def refresh(identifier_type, identifier): @returns_problem_detail @requires_admin @requires_csrf_token -def edit_classifications(identifier_type, identifier): +def edit_classifications(identifier_type: str, identifier: str) -> Any: return app.manager.admin_work_controller.edit_classifications( identifier_type, identifier ) @@ -281,25 +287,25 @@ def edit_classifications(identifier_type, identifier): @app.route("/admin/roles") @returns_json_or_response_or_problem_detail -def roles(): +def roles() -> Any: return app.manager.admin_work_controller.roles() @app.route("/admin/languages") @returns_json_or_response_or_problem_detail -def languages(): +def languages() -> Any: return app.manager.admin_work_controller.languages() @app.route("/admin/media") @returns_json_or_response_or_problem_detail -def media(): +def media() -> Any: return app.manager.admin_work_controller.media() @app.route("/admin/rights_status") @returns_json_or_response_or_problem_detail -def rights_status(): +def rights_status() -> Any: return app.manager.admin_work_controller.rights_status() @@ -307,7 +313,7 @@ def rights_status(): @has_library @returns_problem_detail @requires_admin -def suppressed(): +def suppressed() -> Any: """Returns a feed of suppressed works.""" return app.manager.admin_feed_controller.suppressed() @@ -316,7 +322,7 @@ def suppressed(): @has_library @returns_json_or_response_or_problem_detail @requires_admin -def suppressed_search(): +def suppressed_search() -> Any: """Search within suppressed/hidden works.""" return app.manager.admin_feed_controller.suppressed_search() @@ -324,7 +330,7 @@ def suppressed_search(): @app.route("/admin/genres") @returns_json_or_response_or_problem_detail @requires_admin -def genres(): +def genres() -> Any: """Returns a JSON representation of complete genre tree.""" return app.manager.admin_feed_controller.genres() @@ -333,7 +339,7 @@ def genres(): @returns_problem_detail @allows_library @requires_admin -def bulk_circulation_events(): +def bulk_circulation_events() -> Any: """Returns a CSV representation of all circulation events with optional start and end times.""" ( @@ -361,7 +367,7 @@ def bulk_circulation_events(): @app.route("/admin/stats") @returns_json_or_response_or_problem_detail @requires_admin -def stats(): +def stats() -> Any: statistics_response: StatisticsResponse = ( app.manager.admin_dashboard_controller.stats(stats_function=generate_statistics) ) @@ -371,7 +377,7 @@ def stats(): @app.route("/admin/quicksight_embed/") @returns_json_or_response_or_problem_detail @requires_admin -def generate_quicksight_url(dashboard_name: str): +def generate_quicksight_url(dashboard_name: str) -> Any: return app.manager.admin_quicksight_controller.generate_quicksight_url( dashboard_name ) @@ -380,7 +386,7 @@ def generate_quicksight_url(dashboard_name: str): @app.route("/admin/quicksight_embed/names") @returns_json_or_response_or_problem_detail @requires_admin -def get_quicksight_names(): +def get_quicksight_names() -> Any: return app.manager.admin_quicksight_controller.get_dashboard_names() @@ -388,7 +394,7 @@ def get_quicksight_names(): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def libraries(): +def libraries() -> Any: return app.manager.admin_library_settings_controller.process_libraries() @@ -396,7 +402,7 @@ def libraries(): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def library(library_uuid): +def library(library_uuid: str) -> Any: return app.manager.admin_library_settings_controller.process_delete(library_uuid) @@ -404,7 +410,7 @@ def library(library_uuid): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def collections(): +def collections() -> Any: return app.manager.admin_collection_settings_controller.process_collections() @@ -412,7 +418,7 @@ def collections(): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def collection(collection_id): +def collection(collection_id: str) -> Any: return app.manager.admin_collection_settings_controller.process_delete( collection_id ) @@ -422,7 +428,7 @@ def collection(collection_id): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def collection_self_tests(identifier): +def collection_self_tests(identifier: str) -> Any: return ( app.manager.admin_collection_settings_controller.process_collection_self_tests( identifier @@ -435,7 +441,7 @@ def collection_self_tests(identifier): @allows_admin_auth_setup @requires_admin @requires_csrf_token -def individual_admins(): +def individual_admins() -> Any: return ( app.manager.admin_individual_admin_settings_controller.process_individual_admins() ) @@ -445,7 +451,7 @@ def individual_admins(): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def individual_admin(email): +def individual_admin(email: str) -> Any: return app.manager.admin_individual_admin_settings_controller.process_delete(email) @@ -453,7 +459,7 @@ def individual_admin(email): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def patron_auth_services(): +def patron_auth_services() -> Any: return ( app.manager.admin_patron_auth_services_controller.process_patron_auth_services() ) @@ -463,7 +469,7 @@ def patron_auth_services(): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def patron_auth_service(service_id): +def patron_auth_service(service_id: str) -> Any: return app.manager.admin_patron_auth_services_controller.process_delete(service_id) @@ -473,7 +479,7 @@ def patron_auth_service(service_id): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def patron_auth_self_tests(identifier): +def patron_auth_self_tests(identifier: str) -> Any: return app.manager.admin_patron_auth_services_controller.process_patron_auth_service_self_tests( identifier ) @@ -484,7 +490,7 @@ def patron_auth_self_tests(identifier): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def lookup_patron(): +def lookup_patron() -> Any: return app.manager.admin_patron_controller.lookup_patron() @@ -493,7 +499,7 @@ def lookup_patron(): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def reset_adobe_id(): +def reset_adobe_id() -> Any: return app.manager.admin_patron_controller.reset_adobe_id() @@ -501,7 +507,7 @@ def reset_adobe_id(): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def metadata_services(): +def metadata_services() -> Any: return app.manager.admin_metadata_services_controller.process_metadata_services() @@ -509,7 +515,7 @@ def metadata_services(): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def metadata_service(service_id): +def metadata_service(service_id: str) -> Any: return app.manager.admin_metadata_services_controller.process_delete(service_id) @@ -517,7 +523,7 @@ def metadata_service(service_id): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def metadata_service_self_tests(identifier): +def metadata_service_self_tests(identifier: str) -> Any: return app.manager.admin_metadata_services_controller.process_metadata_service_self_tests( identifier ) @@ -527,7 +533,7 @@ def metadata_service_self_tests(identifier): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def catalog_services(): +def catalog_services() -> Any: return app.manager.admin_catalog_services_controller.process_catalog_services() @@ -535,7 +541,7 @@ def catalog_services(): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def catalog_service(service_id): +def catalog_service(service_id: str) -> Any: return app.manager.admin_catalog_services_controller.process_delete(service_id) @@ -543,7 +549,7 @@ def catalog_service(service_id): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def discovery_services(): +def discovery_services() -> Any: return app.manager.admin_discovery_services_controller.process_discovery_services() @@ -551,7 +557,7 @@ def discovery_services(): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def discovery_service(service_id): +def discovery_service(service_id: str) -> Any: return app.manager.admin_discovery_services_controller.process_delete(service_id) @@ -559,7 +565,7 @@ def discovery_service(service_id): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def announcements_for_all(): +def announcements_for_all() -> Any: return app.manager.admin_announcement_service.process_many() @@ -567,7 +573,7 @@ def announcements_for_all(): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def discovery_service_library_registrations(): +def discovery_service_library_registrations() -> Any: return ( app.manager.admin_discovery_service_library_registrations_controller.process_discovery_service_library_registrations() ) @@ -578,7 +584,7 @@ def discovery_service_library_registrations(): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def custom_lists_post(): +def custom_lists_post() -> Any: return app.manager.admin_custom_lists_controller.custom_lists() @@ -587,7 +593,7 @@ def custom_lists_post(): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def custom_lists_get(): +def custom_lists_get() -> Any: return app.manager.admin_custom_lists_controller.custom_lists() @@ -596,7 +602,7 @@ def custom_lists_get(): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def custom_list_get(list_id: int): +def custom_list_get(list_id: str) -> Any: return app.manager.admin_custom_lists_controller.custom_list(list_id) @@ -605,7 +611,7 @@ def custom_list_get(list_id: int): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def custom_list_post(list_id): +def custom_list_post(list_id: str) -> Any: return app.manager.admin_custom_lists_controller.custom_list(list_id) @@ -614,7 +620,7 @@ def custom_list_post(list_id): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def custom_list_delete(list_id): +def custom_list_delete(list_id: str) -> Any: return app.manager.admin_custom_lists_controller.custom_list(list_id) @@ -623,7 +629,7 @@ def custom_list_delete(list_id): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def custom_list_share(list_id: int): +def custom_list_share(list_id: str) -> Any: """Share a custom list with all libraries in the CM that share the collections of this library and works of this list""" return app.manager.admin_custom_lists_controller.share_locally(list_id) @@ -633,7 +639,7 @@ def custom_list_share(list_id: int): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def custom_list_unshare(list_id: int): +def custom_list_unshare(list_id: str) -> Any: """Unshare the list from all libraries, as long as no other library is using the list in its lanes""" return app.manager.admin_custom_lists_controller.share_locally(list_id) @@ -643,7 +649,7 @@ def custom_list_unshare(list_id: int): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def lanes(): +def lanes() -> Any: return app.manager.admin_lanes_controller.lanes() @@ -652,7 +658,7 @@ def lanes(): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def lane(lane_identifier): +def lane(lane_identifier: str) -> Any: return app.manager.admin_lanes_controller.lane(lane_identifier) @@ -661,7 +667,7 @@ def lane(lane_identifier): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def lane_show(lane_identifier): +def lane_show(lane_identifier: str) -> Any: return app.manager.admin_lanes_controller.show_lane(lane_identifier) @@ -670,7 +676,7 @@ def lane_show(lane_identifier): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def lane_hide(lane_identifier): +def lane_hide(lane_identifier: str) -> Any: return app.manager.admin_lanes_controller.hide_lane(lane_identifier) @@ -679,7 +685,7 @@ def lane_hide(lane_identifier): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def reset_lanes(): +def reset_lanes() -> Any: return app.manager.admin_lanes_controller.reset() @@ -688,7 +694,7 @@ def reset_lanes(): @returns_json_or_response_or_problem_detail @requires_admin @requires_csrf_token -def change_lane_order(): +def change_lane_order() -> Any: return app.manager.admin_lanes_controller.change_order() @@ -696,14 +702,14 @@ def change_lane_order(): @has_library @returns_json_or_response_or_problem_detail @requires_admin -def search_field_values(): +def search_field_values() -> Any: return app.manager.admin_search_controller.search_field_values() @app.route("/admin/diagnostics") @requires_admin @returns_json_or_response_or_problem_detail -def diagnostics(): +def diagnostics() -> Any: return app.manager.timestamps_controller.diagnostics() @@ -714,7 +720,7 @@ def diagnostics(): @allows_library @returns_json_or_response_or_problem_detail @requires_admin -def inventory_report_info(): +def inventory_report_info() -> Any: return app.manager.admin_report_controller.inventory_report_info() @@ -725,7 +731,7 @@ def inventory_report_info(): @allows_library @returns_json_or_response_or_problem_detail @requires_admin -def generate_inventory_report(): +def generate_inventory_report() -> Any: return app.manager.admin_report_controller.generate_inventory_report() @@ -733,21 +739,22 @@ def generate_inventory_report(): @has_library @returns_json_or_response_or_problem_detail @requires_admin -def generate_report(report_key: str): +def generate_report(report_key: str) -> Any: return app.manager.admin_report_controller.generate_report(report_key=report_key) @app.route("/admin/sign_in_again") -def admin_sign_in_again(): +def admin_sign_in_again() -> Any: """Allows an admin with expired credentials to sign back in from a new browser tab so they won't lose changes. """ admin = app.manager.admin_sign_in_controller.authenticated_admin_from_request() csrf_token = app.manager.admin_sign_in_controller.get_csrf_token() + # Mock in tests can make get_csrf_token() return ProblemDetail if ( isinstance(admin, ProblemDetail) or csrf_token is None - or isinstance(csrf_token, ProblemDetail) + or isinstance(csrf_token, ProblemDetail) # type: ignore[unreachable] ): redirect_url = flask.request.url return redirect(url_for("admin_sign_in", redirect=redirect_url, _external=True)) @@ -761,19 +768,24 @@ def admin_sign_in_again(): @app.route("/admin/web/collection/") @app.route("/admin/web/book/") @app.route("/admin/web/") # catchall for single-page URLs -def admin_view(collection=None, book=None, etc=None, **kwargs): +def admin_view( + collection: str | None = None, + book: str | None = None, + etc: str | None = None, + **kwargs: Any, +) -> Any: return app.manager.admin_view_controller(collection, book, path=etc) @app.route("/admin/", strict_slashes=False) -def admin_base(**kwargs): +def admin_base(**kwargs: Any) -> Any: return redirect(url_for("admin_view", _external=True)) @app.route("/admin/libraries/import", strict_slashes=False, methods=["POST"]) @returns_json_or_response_or_problem_detail @requires_basic_auth -def import_libraries(): +def import_libraries() -> Any: """Import multiple libraries from a list of library configurations.""" return app.manager.admin_library_settings_controller.import_libraries() @@ -783,7 +795,7 @@ def import_libraries(): @app.route("/admin/static/") @returns_problem_detail - def admin_static_file(filename): + def admin_static_file(filename: str) -> Any: return StaticFileController.static_file( AdminClientConfig.static_files_directory(), filename ) diff --git a/src/palace/manager/api/admin/template_styles.py b/src/palace/manager/api/admin/template_styles.py index 3d90932031..eaac1b0425 100644 --- a/src/palace/manager/api/admin/template_styles.py +++ b/src/palace/manager/api/admin/template_styles.py @@ -1,4 +1,4 @@ -body_style = """ +body_style: str = """ margin: 10vh auto; font-family: 'Open Sans',Helvetica,Arial,sans-serif; padding: 25px 15px; @@ -11,17 +11,17 @@ align-items: center; """ -label_style = """ +label_style: str = """ font-weight: 700; """ -error_style = ( +error_style: str = ( body_style + """ border-color: #D0343A; """ ) -input_style = """ +input_style: str = """ border-radius: .25em; display: block; padding: 10px; @@ -32,14 +32,14 @@ width: 25vw; """ -section_style = """ +section_style: str = """ width: 25vw; padding: 12px; display: flex; justify-content: space-between; align-items: center; """ -button_style = """ +button_style: str = """ background: #242DAB; border-color: transparent; border-radius: .25em; @@ -53,7 +53,7 @@ margin: 2vh auto; """ -link_style = """ +link_style: str = """ background: #242DAB; text-align: center; text-decoration: none; @@ -68,7 +68,7 @@ margin: 2vh auto; """ -small_link_style = ( +small_link_style: str = ( link_style + """ width: 5vw; @@ -76,14 +76,14 @@ """ ) -hr_style = """ +hr_style: str = """ width: 10vw; margin: 3px 0 0 0; border: none; border-bottom: 1px solid #403d37; """ -logo_style = """ +logo_style: str = """ width: 200px; margin: 20px; """ diff --git a/src/palace/manager/api/admin/validator.py b/src/palace/manager/api/admin/validator.py index 7a024c143e..565dc4b059 100644 --- a/src/palace/manager/api/admin/validator.py +++ b/src/palace/manager/api/admin/validator.py @@ -1,4 +1,10 @@ +from __future__ import annotations + +# mypy: warn_unreachable=false import re +from collections.abc import Callable +from re import Match +from typing import Any from flask_babel import lazy_gettext as _ @@ -7,11 +13,14 @@ INVALID_NUMBER, INVALID_URL, ) +from palace.manager.util.problem_detail import ProblemDetail class Validator: - def validate(self, settings, content): - validators = [ + def validate( + self, settings: list[dict[str, Any]] | str, content: dict[str, Any] + ) -> ProblemDetail | None: + validators: list[Callable[..., ProblemDetail | None],] = [ self.validate_email, self.validate_url, self.validate_number, @@ -21,12 +30,20 @@ def validate(self, settings, content): error = validator(settings, content) if error: return error + return None def _extract_inputs( - self, settings, value, form, key="format", is_list=False, should_zip=False - ): + self, + settings: list[dict[str, Any]], + value: str, + form: dict[str, Any] | None, + key: str = "format", + is_list: bool = False, + should_zip: bool = False, + ) -> list[Any]: if not (isinstance(settings, list)): return [] + form = form or {} fields = [s for s in settings if s.get(key) == value and self._value(s, form)] @@ -40,7 +57,9 @@ def _extract_inputs( else: return values - def validate_email(self, settings, content): + def validate_email( + self, settings: list[dict[str, Any]] | str, content: dict[str, Any] + ) -> ProblemDetail | None: """Find any email addresses that the user has submitted, and make sure that they are in a valid format. This method is used by individual_admin_settings and library_settings. @@ -49,7 +68,9 @@ def validate_email(self, settings, content): # If :param settings is a list of objects--i.e. the LibrarySettingsController # is calling this method--then we need to pull out the relevant input strings # to validate. - email_inputs = self._extract_inputs(settings, "email", content.get("form")) + email_inputs = self._extract_inputs( + settings, "email", content.get("form") or {} + ) else: # If the IndividualAdminSettingsController is calling this method, then we already have the # input string; it was passed in directly. @@ -64,15 +85,20 @@ def validate_email(self, settings, content): return INVALID_EMAIL.detailed( _('"%(email)s" is not a valid email address.', email=email) ) + return None - def _is_email(self, email): + def _is_email(self, email: str) -> Match[str] | None: """Email addresses must be in the format 'x@y.z'.""" email_format = r".+\@.+\..+" return re.search(email_format, email) - def validate_url(self, settings, content): + def validate_url( + self, settings: list[dict[str, Any]] | str, content: dict[str, Any] + ) -> ProblemDetail | None: """Find any URLs that the user has submitted, and make sure that they are in a valid format.""" + if not isinstance(settings, list): + return None # Find the fields that have to do with URLs and are not blank. url_inputs = self._extract_inputs( settings, "url", content.get("form"), should_zip=True @@ -89,9 +115,10 @@ def validate_url(self, settings, content): return INVALID_URL.detailed( _('"%(url)s" is not a valid URL.', url=url) ) + return None @classmethod - def _is_url(cls, url, allowed): + def _is_url(cls, url: str, allowed: list[str]) -> bool: if not url: return False has_protocol = any( @@ -99,9 +126,13 @@ def _is_url(cls, url, allowed): ) return has_protocol or (url in allowed) - def validate_number(self, settings, content): + def validate_number( + self, settings: list[dict[str, Any]] | str, content: dict[str, Any] + ) -> ProblemDetail | None: """Find any numbers that the user has submitted, and make sure that they are 1) actually numbers, 2) positive, and 3) lower than the specified maximum, if there is one.""" + if not isinstance(settings, list): + return None # Find the fields that should have numeric input and are not blank. number_inputs = self._extract_inputs( settings, "number", content.get("form"), key="type", should_zip=True @@ -110,8 +141,9 @@ def validate_number(self, settings, content): error = self._number_error(field, number) if error: return error + return None - def _number_error(self, field, number): + def _number_error(self, field: dict[str, Any], number: Any) -> ProblemDetail | None: min = field.get("min") or 0 max = field.get("max") @@ -138,21 +170,28 @@ def _number_error(self, field, number): max=max, ) ) + return None - def _list_of_values(self, fields, form): - result = [] + def _list_of_values(self, fields: list[dict[str, Any]], form: Any) -> list[Any]: + result: list[Any] = [] for field in fields: result += self._value(field, form) return [_f for _f in result if _f] - def _value(self, field, form): + def _value(self, field: dict[str, Any], form: Any) -> Any: # Extract the user's input for this field. If this is a sitewide setting, # then the input needs to be accessed via "value" rather than via the setting's key. # We use getlist instead of get so that, if the field is such that the user can input multiple values # (e.g. language codes), we'll extract all the values, not just the first one. - value = form.getlist(field.get("key")) + # form is typed as Any because it may be ImmutableMultiDict (has getlist) or dict-like. + getlist = getattr(form, "getlist", None) + if getlist is not None: + value = list(getlist(field.get("key"))) + else: + v = form.get(field.get("key")) + value = [v] if v is not None else [] if not value: return form.get("value") - elif len(value) == 1: + if len(value) == 1: return value[0] - return [x for x in value if x != None and x != ""] + return [x for x in value if x is not None and x != ""]