Helper function to get flask.request.library (#2177)

* Type hint flask.request.library

* Add some documentation

* Code review feedback

pyproject.toml

@@ -98,6 +98,7 @@ module = [
     "palace.manager.api.metadata.*",
     "palace.manager.api.odl.*",
     "palace.manager.api.opds_for_distributors",
+    "palace.manager.api.util.flask",
     "palace.manager.core.opds2_import",
     "palace.manager.core.opds_import",
     "palace.manager.core.opds_schema",

src/palace/manager/api/admin/controller/admin_search.py

@@ -1,9 +1,9 @@
 from __future__ import annotations
 
-import flask
 from sqlalchemy import func, or_
 
 from palace.manager.api.admin.controller.base import AdminController
+from palace.manager.api.util.flask import get_request_library
 from palace.manager.sqlalchemy.model.classification import (
     Classification,
     Genre,
@@ -30,7 +30,7 @@ def search_field_values(self) -> dict:
         - Publisher
         - Subject
         """
-        library: Library = flask.request.library  # type: ignore
+        library = get_request_library()
         collection_ids = [coll.id for coll in library.collections if coll.id]
         return self._search_field_values_cached(collection_ids)
 

src/palace/manager/api/admin/controller/custom_lists.py

@@ -23,6 +23,7 @@
     CirculationManagerController,
 )
 from palace.manager.api.problem_details import CANNOT_DELETE_SHARED_LIST
+from palace.manager.api.util.flask import get_request_library
 from palace.manager.core.app_server import load_pagination_from_request
 from palace.manager.core.problem_details import INVALID_INPUT, METHOD_NOT_ALLOWED
 from palace.manager.core.query.customlist import CustomListQueries
@@ -83,7 +84,7 @@ def _list_as_json(self, list: CustomList, is_owner=True) -> dict:
         )
 
     def custom_lists(self) -> dict | ProblemDetail | Response | None:
-        library: Library = flask.request.library  # type: ignore  # "Request" has no attribute "library"
+        library = get_request_library()
         self.require_librarian(library)
 
         if flask.request.method == "GET":
@@ -322,7 +323,7 @@ def url_fn(after):
         return url_fn
 
     def custom_list(self, list_id: int) -> Response | dict | ProblemDetail | None:
-        library: Library = flask.request.library  # type: ignore
+        library = get_request_library()
         self.require_librarian(library)
         data_source = DataSource.lookup(self._db, DataSource.LIBRARY_STAFF)
 
@@ -375,7 +376,7 @@ def custom_list(self, list_id: int) -> Response | dict | ProblemDetail | None:
 
         elif flask.request.method == "DELETE":
             # Deleting requires a library manager.
-            self.require_library_manager(flask.request.library)  # type: ignore
+            self.require_library_manager(get_request_library())
 
             if len(list.shared_locally_with_libraries) > 0:
                 return CANNOT_DELETE_SHARED_LIST
@@ -413,7 +414,7 @@ def share_locally(
         customlist = get_one(self._db, CustomList, id=customlist_id)
         if not customlist:
             return MISSING_CUSTOM_LIST
-        if customlist.library != flask.request.library:  # type: ignore
+        if customlist.library != get_request_library():
             return ADMIN_NOT_AUTHORIZED.detailed(
                 _("This library does not have permissions on this customlist.")
             )

src/palace/manager/api/admin/controller/dashboard.py

@@ -12,6 +12,7 @@
     CirculationManagerController,
 )
 from palace.manager.api.local_analytics_exporter import LocalAnalyticsExporter
+from palace.manager.api.util.flask import get_request_library
 from palace.manager.feed.annotator.admin import AdminAnnotator
 from palace.manager.sqlalchemy.model.admin import Admin
 from palace.manager.sqlalchemy.model.circulationevent import CirculationEvent
@@ -29,7 +30,7 @@ def stats(
         return stats_function(admin, self._db)
 
     def circulation_events(self):
-        annotator = AdminAnnotator(self.circulation, flask.request.library)
+        annotator = AdminAnnotator(self.circulation, get_request_library())
         num = min(int(flask.request.args.get("num", "100")), 500)
 
         results = (
@@ -92,7 +93,7 @@ def get_date(field):
         date_end_label = get_date("dateEnd")
         date_end = date_end_label + timedelta(days=1)
         locations = flask.request.args.get("locations", None)
-        library = getattr(flask.request, "library", None)
+        library = get_request_library(default=None)
         library_short_name = library.short_name if library else None
 
         analytics_exporter = analytics_exporter or LocalAnalyticsExporter()

src/palace/manager/api/admin/controller/lanes.py

@@ -20,6 +20,7 @@
     CirculationManagerController,
 )
 from palace.manager.api.lanes import create_default_lanes
+from palace.manager.api.util.flask import get_request_library
 from palace.manager.sqlalchemy.model.customlist import CustomList
 from palace.manager.sqlalchemy.model.lane import Lane
 from palace.manager.sqlalchemy.model.library import Library
@@ -28,7 +29,7 @@
 
 class LanesController(CirculationManagerController, AdminPermissionsControllerMixin):
     def lanes(self):
-        library = flask.request.library
+        library = get_request_library()
         self.require_librarian(library)
 
         if flask.request.method == "GET":
@@ -56,7 +57,7 @@ def lanes_for_parent(parent):
             return dict(lanes=lanes_for_parent(None))
 
         if flask.request.method == "POST":
-            self.require_library_manager(flask.request.library)
+            self.require_library_manager(get_request_library())
 
             id = flask.request.form.get("id")
             parent_id = flask.request.form.get("parent_id")
@@ -173,7 +174,7 @@ def lanes_for_parent(parent):
 
     def lane(self, lane_identifier):
         if flask.request.method == "DELETE":
-            library = flask.request.library
+            library = get_request_library()
             self.require_library_manager(library)
 
             lane = get_one(self._db, Lane, id=lane_identifier, library=library)
@@ -192,7 +193,7 @@ def delete_lane_and_sublanes(lane):
             return Response(str(_("Deleted")), 200)
 
     def show_lane(self, lane_identifier):
-        library = flask.request.library
+        library = get_request_library()
         self.require_library_manager(library)
 
         lane = get_one(self._db, Lane, id=lane_identifier, library=library)
@@ -204,7 +205,7 @@ def show_lane(self, lane_identifier):
         return Response(str(_("Success")), 200)
 
     def hide_lane(self, lane_identifier):
-        library = flask.request.library
+        library = get_request_library()
         self.require_library_manager(library)
 
         lane = get_one(self._db, Lane, id=lane_identifier, library=library)
@@ -214,13 +215,14 @@ def hide_lane(self, lane_identifier):
         return Response(str(_("Success")), 200)
 
     def reset(self):
-        self.require_library_manager(flask.request.library)
+        library = get_request_library()
+        self.require_library_manager(library)
 
-        create_default_lanes(self._db, flask.request.library)
+        create_default_lanes(self._db, library)
         return Response(str(_("Success")), 200)
 
     def change_order(self):
-        self.require_library_manager(flask.request.library)
+        self.require_library_manager(get_request_library())
 
         submitted_lanes = json.loads(flask.request.data)
 

src/palace/manager/api/admin/controller/work_editor.py

@@ -27,6 +27,7 @@
     LIBRARY_NOT_FOUND,
     REMOTE_INTEGRATION_FAILED,
 )
+from palace.manager.api.util.flask import get_request_library
 from palace.manager.core.classifier import NO_NUMBER, NO_VALUE, genres
 from palace.manager.core.classifier.simplified import SimplifiedGenreClassifier
 from palace.manager.feed.acquisition import OPDSAcquisitionFeed
@@ -62,13 +63,14 @@ def details(self, identifier_type, identifier):
 
         :return: An OPDSEntryResponse
         """
-        self.require_librarian(flask.request.library)
+        library = get_request_library()
+        self.require_librarian(library)
 
-        work = self.load_work(flask.request.library, identifier_type, identifier)
+        work = self.load_work(library, identifier_type, identifier)
         if isinstance(work, ProblemDetail):
             return work
 
-        annotator = AdminAnnotator(self.circulation, flask.request.library)
+        annotator = AdminAnnotator(self.circulation, library)
 
         # single_entry returns an OPDSEntryResponse that will not be
         # cached, which is perfect. We want the admin interface
@@ -136,7 +138,8 @@ def rights_status(self):
 
     def edit(self, identifier_type, identifier):
         """Edit a work's metadata."""
-        self.require_librarian(flask.request.library)
+        library = get_request_library()
+        self.require_librarian(library)
 
         # TODO: It would be nice to use the metadata layer for this, but
         # this code handles empty values differently than other metadata
@@ -145,7 +148,7 @@ def edit(self, identifier_type, identifier):
         # db so that it can overrule other data sources that set a value,
         # unlike other sources which set empty fields to None.
 
-        work = self.load_work(flask.request.library, identifier_type, identifier)
+        work = self.load_work(library, identifier_type, identifier)
         if isinstance(work, ProblemDetail):
             return work
 
@@ -372,7 +375,7 @@ def suppress(
     ) -> Response | ProblemDetail:
         """Suppress a book at the level of a library."""
 
-        library: Library | None = getattr(flask.request, "library")
+        library: Library | None = get_request_library(default=None)
         if library is None:
             raise ProblemDetailException(LIBRARY_NOT_FOUND)
 
@@ -404,7 +407,7 @@ def unsuppress(
     ) -> Response | ProblemDetail:
         """Remove a book suppression from a book at the level of a library"""
 
-        library: Library | None = getattr(flask.request, "library")
+        library: Library | None = get_request_library(default=None)
         if library is None:
             raise ProblemDetailException(LIBRARY_NOT_FOUND)
 
@@ -433,9 +436,10 @@ def unsuppress(
 
     def refresh_metadata(self, identifier_type, identifier, provider=None):
         """Refresh the metadata for a book from the content server"""
-        self.require_librarian(flask.request.library)
+        library = get_request_library()
+        self.require_librarian(library)
 
-        work = self.load_work(flask.request.library, identifier_type, identifier)
+        work = self.load_work(library, identifier_type, identifier)
         if isinstance(work, ProblemDetail):
             return work
 
@@ -464,9 +468,10 @@ def refresh_metadata(self, identifier_type, identifier, provider=None):
 
     def classifications(self, identifier_type, identifier):
         """Return list of this work's classifications."""
-        self.require_librarian(flask.request.library)
+        library = get_request_library()
+        self.require_librarian(library)
 
-        work = self.load_work(flask.request.library, identifier_type, identifier)
+        work = self.load_work(library, identifier_type, identifier)
         if isinstance(work, ProblemDetail):
             return work
 
@@ -502,9 +507,10 @@ def classifications(self, identifier_type, identifier):
 
     def edit_classifications(self, identifier_type, identifier):
         """Edit a work's audience, target age, fiction status, and genres."""
-        self.require_librarian(flask.request.library)
+        library = get_request_library()
+        self.require_librarian(library)
 
-        work = self.load_work(flask.request.library, identifier_type, identifier)
+        work = self.load_work(library, identifier_type, identifier)
         if isinstance(work, ProblemDetail):
             return work
 
@@ -670,9 +676,8 @@ def edit_classifications(self, identifier_type, identifier):
         return Response("", 200)
 
     def custom_lists(self, identifier_type, identifier):
-        self.require_librarian(flask.request.library)
-
-        library = flask.request.library
+        library = get_request_library()
+        self.require_librarian(library)
         work = self.load_work(library, identifier_type, identifier)
         if isinstance(work, ProblemDetail):
             return work

src/palace/manager/api/authenticator.py

@@ -33,6 +33,7 @@
     UNKNOWN_SAML_PROVIDER,
     UNSUPPORTED_AUTHENTICATION_MECHANISM,
 )
+from palace.manager.api.util.flask import get_request_library
 from palace.manager.core.user_profile import ProfileController
 from palace.manager.integration.goals import Goals
 from palace.manager.service.analytics.analytics import Analytics
@@ -129,7 +130,7 @@ def __init__(
 
     @property
     def current_library_short_name(self):
-        return flask.request.library.short_name
+        return get_request_library().short_name
 
     def populate_authenticators(
         self, _db, libraries: Iterable[Library], analytics: Analytics | None

src/palace/manager/api/circulation.py

@@ -35,6 +35,7 @@
     PatronHoldLimitReached,
     PatronLoanLimitReached,
 )
+from palace.manager.api.util.flask import get_request_library
 from palace.manager.api.util.patron import PatronUtility
 from palace.manager.core.exceptions import PalaceValueError
 from palace.manager.integration.base import HasLibraryIntegrationConfiguration
@@ -948,12 +949,12 @@ def _collect_event(
         if patron:
             # The library of the patron who caused the event.
             library = patron.library
-        elif flask.request and getattr(flask.request, "library", None):
-            # The library associated with the current request.
-            library = getattr(flask.request, "library")
         else:
-            # The library associated with the CirculationAPI itself.
-            library = self.library
+            # The library associated with the current request, defaulting to
+            # the library associated with the CirculationAPI itself if we are
+            # outside a request context, or if the request context does not
+            # have a library associated with it.
+            library = get_request_library(default=self.library)
 
         neighborhood = None
         if (

src/palace/manager/api/circulation_manager.py

@@ -32,6 +32,7 @@
 from palace.manager.api.lanes import load_lanes
 from palace.manager.api.problem_details import NO_SUCH_LANE
 from palace.manager.api.saml.controller import SAMLController
+from palace.manager.api.util.flask import get_request_library
 from palace.manager.core.app_server import (
     ApplicationVersionController,
     load_facets_from_request,
@@ -392,7 +393,7 @@ def annotator(self, lane, facets=None, *args, **kwargs):
         elif lane and isinstance(lane, WorkList):
             library = lane.get_library(self._db)
         if not library and hasattr(flask.request, "library"):
-            library = flask.request.library
+            library = get_request_library()
 
         # If no library is provided, the best we can do is a generic
         # annotator for this application.
@@ -434,7 +435,7 @@ def authentication_for_opds_document(self):
         internal details of deployment, it should only be enabled when
         diagnosing deployment problems.
         """
-        name = flask.request.library.short_name
+        name = get_request_library().short_name
         value = self.authentication_for_opds_documents.get(name, None)
         if value is None:
             # The document was not in the cache, either because it's

src/palace/manager/api/controller/analytics.py

@@ -7,6 +7,7 @@
     CirculationManagerController,
 )
 from palace.manager.api.problem_details import INVALID_ANALYTICS_EVENT_TYPE
+from palace.manager.api.util.flask import get_request_library
 from palace.manager.sqlalchemy.model.circulationevent import CirculationEvent
 from palace.manager.util.datetime_helpers import utc_now
 from palace.manager.util.problem_detail import ProblemDetail
@@ -18,7 +19,7 @@ def track_event(self, identifier_type, identifier, event_type):
         # a way to distinguish between different LicensePools for the
         # same book.
         if event_type in CirculationEvent.CLIENT_EVENTS:
-            library = flask.request.library
+            library = get_request_library()
             # Authentication on the AnalyticsController is optional,
             # so flask.request.patron may or may not be set.
             patron = getattr(flask.request, "patron", None)

src/palace/manager/api/controller/base.py

@@ -123,5 +123,5 @@ def library_for_request(
 
         if not library:
             return LIBRARY_NOT_FOUND
-        flask.request.library = library  # type: ignore[attr-defined]
+        setattr(flask.request, "library", library)
         return library