From e7c86e59463e120a5caa90c9bc72644f07616ae7 Mon Sep 17 00:00:00 2001 From: Bert Blommers Date: Sat, 6 Jan 2024 08:24:12 +0000 Subject: [PATCH] Techdebt: Simplify PinPoint routing (#7191) --- moto/pinpoint/responses.py | 62 ++++++++++---------------------------- moto/pinpoint/urls.py | 14 +++------ 2 files changed, 20 insertions(+), 56 deletions(-) diff --git a/moto/pinpoint/responses.py b/moto/pinpoint/responses.py index ad7ad4f38..c323fa7a5 100644 --- a/moto/pinpoint/responses.py +++ b/moto/pinpoint/responses.py @@ -20,36 +20,6 @@ class PinpointResponse(BaseResponse): """Return backend instance specific for this region.""" return pinpoint_backends[self.current_account][self.region] - def app(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] - self.setup_class(request, full_url, headers) - if request.method == "DELETE": - return self.delete_app() - if request.method == "GET": - return self.get_app() - - def apps(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] - self.setup_class(request, full_url, headers) - if request.method == "GET": - return self.get_apps() - if request.method == "POST": - return self.create_app() - - def app_settings(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] - self.setup_class(request, full_url, headers) - if request.method == "GET": - return self.get_application_settings() - if request.method == "PUT": - return self.update_application_settings() - - def eventstream(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] - self.setup_class(request, full_url, headers) - if request.method == "DELETE": - return self.delete_event_stream() - if request.method == "GET": - return self.get_event_stream() - if request.method == "POST": - return self.put_event_stream() - def tags(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] self.setup_class(request, full_url, headers) if request.method == "DELETE": @@ -66,22 +36,22 @@ class PinpointResponse(BaseResponse): app = self.pinpoint_backend.create_app(name=name, tags=tags) return 201, {}, json.dumps(app.to_json()) - def delete_app(self) -> TYPE_RESPONSE: + def delete_app(self) -> str: application_id = self.path.split("/")[-1] app = self.pinpoint_backend.delete_app(application_id=application_id) - return 200, {}, json.dumps(app.to_json()) + return json.dumps(app.to_json()) - def get_app(self) -> TYPE_RESPONSE: + def get_app(self) -> str: application_id = self.path.split("/")[-1] app = self.pinpoint_backend.get_app(application_id=application_id) - return 200, {}, json.dumps(app.to_json()) + return json.dumps(app.to_json()) - def get_apps(self) -> TYPE_RESPONSE: + def get_apps(self) -> str: apps = self.pinpoint_backend.get_apps() resp = {"Item": [a.to_json() for a in apps]} - return 200, {}, json.dumps(resp) + return json.dumps(resp) - def update_application_settings(self) -> TYPE_RESPONSE: + def update_application_settings(self) -> str: application_id = self.path.split("/")[-2] settings = json.loads(self.body) app_settings = self.pinpoint_backend.update_application_settings( @@ -89,16 +59,16 @@ class PinpointResponse(BaseResponse): ) response = app_settings.to_json() response["ApplicationId"] = application_id - return 200, {}, json.dumps(response) + return json.dumps(response) - def get_application_settings(self) -> TYPE_RESPONSE: + def get_application_settings(self) -> str: application_id = self.path.split("/")[-2] app_settings = self.pinpoint_backend.get_application_settings( application_id=application_id ) response = app_settings.to_json() response["ApplicationId"] = application_id - return 200, {}, json.dumps(response) + return json.dumps(response) def list_tags_for_resource(self) -> TYPE_RESPONSE: resource_arn = unquote(self.path).split("/tags/")[-1] @@ -119,7 +89,7 @@ class PinpointResponse(BaseResponse): ) return 200, {}, "{}" - def put_event_stream(self) -> TYPE_RESPONSE: + def put_event_stream(self) -> str: application_id = self.path.split("/")[-2] params = json.loads(self.body) stream_arn = params.get("DestinationStreamArn") @@ -129,22 +99,22 @@ class PinpointResponse(BaseResponse): ) resp = event_stream.to_json() resp["ApplicationId"] = application_id - return 200, {}, json.dumps(resp) + return json.dumps(resp) - def get_event_stream(self) -> TYPE_RESPONSE: + def get_event_stream(self) -> str: application_id = self.path.split("/")[-2] event_stream = self.pinpoint_backend.get_event_stream( application_id=application_id ) resp = event_stream.to_json() resp["ApplicationId"] = application_id - return 200, {}, json.dumps(resp) + return json.dumps(resp) - def delete_event_stream(self) -> TYPE_RESPONSE: + def delete_event_stream(self) -> str: application_id = self.path.split("/")[-2] event_stream = self.pinpoint_backend.delete_event_stream( application_id=application_id ) resp = event_stream.to_json() resp["ApplicationId"] = application_id - return 200, {}, json.dumps(resp) + return json.dumps(resp) diff --git a/moto/pinpoint/urls.py b/moto/pinpoint/urls.py index 9d92093f8..4963a3d20 100644 --- a/moto/pinpoint/urls.py +++ b/moto/pinpoint/urls.py @@ -7,16 +7,10 @@ url_bases = [ url_paths = { - "{0}/v1/apps$": PinpointResponse.method_dispatch(PinpointResponse.apps), - "{0}/v1/apps/(?P[^/]+)$": PinpointResponse.method_dispatch( - PinpointResponse.app - ), - "{0}/v1/apps/(?P[^/]+)/eventstream": PinpointResponse.method_dispatch( - PinpointResponse.eventstream - ), - "{0}/v1/apps/(?P[^/]+)/settings$": PinpointResponse.method_dispatch( - PinpointResponse.app_settings - ), + "{0}/v1/apps$": PinpointResponse.dispatch, + "{0}/v1/apps/(?P[^/]+)$": PinpointResponse.dispatch, + "{0}/v1/apps/(?P[^/]+)/eventstream": PinpointResponse.dispatch, + "{0}/v1/apps/(?P[^/]+)/settings$": PinpointResponse.dispatch, "{0}/v1/tags/(?P[^/]+)$": PinpointResponse.method_dispatch( PinpointResponse.tags ),