Refactor GenericModel for reusability

client/src/api/notifications.ts

@@ -25,7 +25,7 @@ export type NewSharedItemNotificationContentItemType =
 
 type UserNotificationUpdateRequest = components["schemas"]["UserNotificationUpdateRequest"];
 
-export type NotificationCreateRequest = components["schemas"]["NotificationCreateRequestBody"];
+export type NotificationCreateRequest = components["schemas"]["NotificationCreateRequest"];
 
 type NotificationResponse = components["schemas"]["NotificationResponse"];
 

client/src/api/schema/schema.ts

@@ -9638,14 +9638,18 @@ export interface components {
              */
             variant: components["schemas"]["NotificationVariant"];
         };
-        /** NotificationCreateRequestBody */
-        NotificationCreateRequestBody: {
+        /** NotificationCreateRequest */
+        NotificationCreateRequest: {
             /**
              * Notification
              * @description The notification to create. The structure depends on the category.
              */
             notification: components["schemas"]["NotificationCreateData"];
-            recipients: components["schemas"]["NotificationRecipientsPayload"];
+            /**
+             * Recipients
+             * @description The recipients of the notification. Can be a combination of users, groups and roles.
+             */
+            recipients: components["schemas"]["NotificationRecipientsRequest"];
         };
         /** NotificationCreatedResponse */
         NotificationCreatedResponse: {
@@ -9660,8 +9664,8 @@ export interface components {
              */
             total_notifications_sent: number;
         };
-        /** NotificationRecipientsPayload */
-        NotificationRecipientsPayload: {
+        /** NotificationRecipientsRequest */
+        NotificationRecipientsRequest: {
             /**
              * Group IDs
              * @description The list of encoded group IDs of the groups that should receive the notification.
@@ -19976,7 +19980,7 @@ export interface operations {
         };
         requestBody: {
             content: {
-                "application/json": components["schemas"]["NotificationCreateRequestBody"];
+                "application/json": components["schemas"]["NotificationCreateRequest"];
             };
         };
         responses: {

lib/galaxy/managers/notification.py

@@ -52,7 +52,7 @@ from galaxy.model.scoped_session import galaxy_scoped_session
 from galaxy.schema.notifications import (
     AnyNotificationContent,
     BroadcastNotificationCreateRequest,
-    GenericNotificationCreateRequest,
+    GenericNotificationCreate,
     MandatoryNotificationCategory,
     MessageNotificationContent,
     NewSharedItemNotificationContent,
@@ -154,9 +154,7 @@ class NotificationManager:
     def can_send_notifications_async(self):
         return self.config.enable_celery_tasks
 
-    def send_notification_to_recipients(
-        self, request: GenericNotificationCreateRequest
-    ) -> Tuple[Optional[Notification], int]:
+    def send_notification_to_recipients(self, request: GenericNotificationCreate) -> Tuple[Optional[Notification], int]:
         """
         Creates a new notification and associates it with all the recipient users.
 

lib/galaxy/schema/generics.py

@@ -0,0 +1,52 @@
+from typing import (
+    Any,
+    Tuple,
+    Type,
+    TypeVar,
+)
+
+from pydantic import BaseModel
+from pydantic.json_schema import GenerateJsonSchema
+
+from galaxy.schema.fields import (
+    DecodedDatabaseIdField,
+    EncodedDatabaseIdField,
+)
+
+DatabaseIdT = TypeVar("DatabaseIdT")
+
+ref_to_name = {}
+
+
+class GenericModel(BaseModel):
+    @classmethod
+    def model_parametrized_name(cls, params: Tuple[Type[Any], ...]) -> str:
+        suffix = cls.__determine_suffix__(params)
+        class_name = cls.__name__.split("Generic", 1)[-1]
+        return f"{class_name}{suffix}"
+
+    @classmethod
+    def __get_pydantic_core_schema__(cls, *args, **kwargs):
+        result = super().__get_pydantic_core_schema__(*args, **kwargs)
+        ref_to_name[result["ref"]] = cls.__name__
+        return result
+
+    @classmethod
+    def __determine_suffix__(cls, params: Tuple[Type[Any], ...]) -> str:
+        suffix = "Incoming"
+        if params[0] is EncodedDatabaseIdField:
+            suffix = "Response"
+        elif params[0] is DecodedDatabaseIdField:
+            suffix = "Request"
+        return suffix
+
+
+class CustomJsonSchema(GenerateJsonSchema):
+    def get_defs_ref(self, core_mode_ref):
+        full_def = super().get_defs_ref(core_mode_ref)
+        choices = self._prioritized_defsref_choices[full_def]
+        ref, mode = core_mode_ref
+        if ref in ref_to_name:
+            for i, choice in enumerate(choices):
+                choices[i] = choice.replace(choices[0], ref_to_name[ref])  # type: ignore[call-overload]
+        return full_def

lib/galaxy/schema/invocation.py

@@ -6,9 +6,6 @@ from typing import (
     Generic,
     List,
     Optional,
-    Tuple,
-    Type,
-    TypeVar,
     Union,
 )
 
@@ -20,7 +17,6 @@ from pydantic import (
     UUID1,
     UUID4,
 )
-from pydantic.json_schema import GenerateJsonSchema
 from typing_extensions import (
     Annotated,
     Literal,
@@ -33,6 +29,10 @@ from galaxy.schema.fields import (
     literal_to_value,
     ModelClassField,
 )
+from galaxy.schema.generics import (
+    DatabaseIdT,
+    GenericModel,
+)
 from galaxy.schema.schema import (
     CreateTimeField,
     DataItemSourceType,
@@ -86,38 +86,10 @@ class CancelReason(str, Enum):
     cancelled_on_review = "cancelled_on_review"
 
 
-DatabaseIdT = TypeVar("DatabaseIdT")
-
-ref_to_name = {}
-
-
-class InvocationMessageBase(BaseModel):
+class InvocationMessageBase(GenericModel):
     reason: Union[CancelReason, FailureReason, WarningReason]
     model_config = ConfigDict(from_attributes=True, populate_by_name=True)
 
-    @classmethod
-    def model_parametrized_name(cls, params: Tuple[Type[Any], ...]) -> str:
-        suffix = "Response" if params[0] is EncodedDatabaseIdField else "Incoming"
-        class_name = cls.__name__.split("Generic", 1)[-1]
-        return f"{class_name}{suffix}"
-
-    @classmethod
-    def __get_pydantic_core_schema__(cls, *args, **kwargs):
-        result = super().__get_pydantic_core_schema__(*args, **kwargs)
-        ref_to_name[result["ref"]] = cls.__name__
-        return result
-
-
-class CustomJsonSchema(GenerateJsonSchema):
-    def get_defs_ref(self, core_mode_ref):
-        full_def = super().get_defs_ref(core_mode_ref)
-        choices = self._prioritized_defsref_choices[full_def]
-        ref, mode = core_mode_ref
-        if ref in ref_to_name:
-            for i, choice in enumerate(choices):
-                choices[i] = choice.replace(choices[0], ref_to_name[ref])  # type: ignore[call-overload]
-        return full_def
-
 
 class GenericInvocationCancellationReviewFailed(InvocationMessageBase, Generic[DatabaseIdT]):
     reason: Literal[CancelReason.cancelled_on_review]

lib/galaxy/schema/notifications.py

@@ -6,7 +6,6 @@ from typing import (
     Generic,
     List,
     Optional,
-    TypeVar,
     Union,
 )
 
@@ -24,6 +23,10 @@ from galaxy.schema.fields import (
     DecodedDatabaseIdField,
     EncodedDatabaseIdField,
 )
+from galaxy.schema.generics import (
+    DatabaseIdT,
+    GenericModel,
+)
 from galaxy.schema.schema import Model
 from galaxy.schema.types import (
     AbsoluteOrRelativeUrl,
@@ -261,10 +264,7 @@ class NotificationCreateData(Model):
     )
 
 
-DatabaseIdT = TypeVar("DatabaseIdT")
-
-
-class GenericNotificationRecipients(Model, Generic[DatabaseIdT]):
+class GenericNotificationRecipients(GenericModel, Generic[DatabaseIdT]):
     """The recipients of a notification. Can be a combination of users, groups and roles."""
 
     user_ids: List[DatabaseIdT] = Field(
@@ -284,7 +284,7 @@ class GenericNotificationRecipients(Model, Generic[DatabaseIdT]):
     )
 
 
-class GenericNotificationCreateRequest(Model, Generic[DatabaseIdT]):
+class GenericNotificationCreate(GenericModel, Generic[DatabaseIdT]):
     """Contains the recipients and the notification to create."""
 
     recipients: GenericNotificationRecipients[DatabaseIdT] = Field(
@@ -299,16 +299,11 @@ class GenericNotificationCreateRequest(Model, Generic[DatabaseIdT]):
     )
 
 
-NotificationCreateRequest = GenericNotificationCreateRequest[int]
+NotificationCreateRequest = GenericNotificationCreate[int]
 NotificationRecipients = GenericNotificationRecipients[int]
 
 
-class NotificationRecipientsPayload(GenericNotificationRecipients[DecodedDatabaseIdField]):
-    pass
-
-
-class NotificationCreateRequestBody(GenericNotificationCreateRequest[DecodedDatabaseIdField]):
-    recipients: NotificationRecipientsPayload
+NotificationCreateRequestBody = GenericNotificationCreate[DecodedDatabaseIdField]
 
 
 class BroadcastNotificationCreateRequest(NotificationCreateData):

lib/galaxy/webapps/galaxy/fast_app.py

@@ -12,7 +12,7 @@ from fastapi.openapi.constants import REF_TEMPLATE
 from starlette.middleware.cors import CORSMiddleware
 from starlette.responses import Response
 
-from galaxy.schema.invocation import CustomJsonSchema
+from galaxy.schema.generics import CustomJsonSchema
 from galaxy.version import VERSION
 from galaxy.webapps.base.api import (
     add_exception_handler,