Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 0 additions & 5 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
12 changes: 10 additions & 2 deletions src/palace/manager/api/admin/admin_authentication_provider.py
Original file line number Diff line number Diff line change
@@ -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()
13 changes: 10 additions & 3 deletions src/palace/manager/api/admin/controller/catalog_services.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
16 changes: 13 additions & 3 deletions src/palace/manager/api/admin/controller/collection_settings.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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:
Expand All @@ -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(
Expand Down
14 changes: 12 additions & 2 deletions src/palace/manager/api/admin/controller/custom_lists.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down
3 changes: 2 additions & 1 deletion src/palace/manager/api/admin/controller/dashboard.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand All @@ -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]:
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this change needed? I don't see a place where LocalAnalyticsExporter.export() returns a ProblemDetail.

date_format = "%Y-%m-%d"

def get_date(field: str) -> date:
Expand Down
13 changes: 10 additions & 3 deletions src/palace/manager/api/admin/controller/discovery_services.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)
Expand Down Expand Up @@ -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
Expand Down
30 changes: 27 additions & 3 deletions src/palace/manager/api/admin/controller/lanes.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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)

Expand All @@ -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)

Expand Down
21 changes: 17 additions & 4 deletions src/palace/manager/api/admin/controller/metadata_services.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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)
17 changes: 14 additions & 3 deletions src/palace/manager/api/admin/controller/patron_auth_services.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand Down
10 changes: 6 additions & 4 deletions src/palace/manager/api/admin/model/custom_lists.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
from typing import Any

from pydantic import NonNegativeInt

from palace.manager.util.flask_util import CustomBaseModel
Expand All @@ -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
Original file line number Diff line number Diff line change
@@ -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,
Expand Down Expand Up @@ -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(
Expand All @@ -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,
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this should be str | None, we don't want to pass a Response into a jinja template do we?

) -> str:
forgot_password_url = url_for("admin_forgot_password")
return render_template(
"admin/auth/forgot-password.html.jinja2",
Expand All @@ -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,
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

) -> str:
reset_password_url = url_for(
"admin_reset_password",
reset_password_token=reset_password_token,
Expand All @@ -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,
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we be more precise about this? Maybe Mapping[str, str] instead of ImmutableMultiDict[str, str] | dict[str, str | 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")
Expand All @@ -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

Expand Down
Loading
Loading