Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Techdebt: Simplify PinPoint routing (#7191)

Browse Source
This commit is contained in:
Bert Blommers 2024-01-06 08:24:12 +00:00 committed by GitHub
parent 57bc607307
commit e7c86e5946
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 20 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
moto/pinpoint/responses.py
View File

@ -20,36 +20,6 @@ class PinpointResponse(BaseResponse):
"""Return backend instance specific for this region.""" """Return backend instance specific for this region."""
return pinpoint_backends[self.current_account][self.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] def tags(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
if request.method == "DELETE": if request.method == "DELETE":
@ -66,22 +36,22 @@ class PinpointResponse(BaseResponse):
app = self.pinpoint_backend.create_app(name=name, tags=tags) app = self.pinpoint_backend.create_app(name=name, tags=tags)
return 201, {}, json.dumps(app.to_json()) return 201, {}, json.dumps(app.to_json())
def delete_app(self) -> TYPE_RESPONSE: def delete_app(self) -> str:
application_id = self.path.split("/")[-1] application_id = self.path.split("/")[-1]
app = self.pinpoint_backend.delete_app(application_id=application_id) 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] application_id = self.path.split("/")[-1]
app = self.pinpoint_backend.get_app(application_id=application_id) 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() apps = self.pinpoint_backend.get_apps()
resp = {"Item": [a.to_json() for a in 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] application_id = self.path.split("/")[-2]
settings = json.loads(self.body) settings = json.loads(self.body)
app_settings = self.pinpoint_backend.update_application_settings( app_settings = self.pinpoint_backend.update_application_settings(
@ -89,16 +59,16 @@ class PinpointResponse(BaseResponse):
) )
response = app_settings.to_json() response = app_settings.to_json()
response["ApplicationId"] = application_id 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] application_id = self.path.split("/")[-2]
app_settings = self.pinpoint_backend.get_application_settings( app_settings = self.pinpoint_backend.get_application_settings(
application_id=application_id application_id=application_id
) )
response = app_settings.to_json() response = app_settings.to_json()
response["ApplicationId"] = application_id response["ApplicationId"] = application_id
return 200, {}, json.dumps(response) return json.dumps(response)
def list_tags_for_resource(self) -> TYPE_RESPONSE: def list_tags_for_resource(self) -> TYPE_RESPONSE:
resource_arn = unquote(self.path).split("/tags/")[-1] resource_arn = unquote(self.path).split("/tags/")[-1]
@ -119,7 +89,7 @@ class PinpointResponse(BaseResponse):
) )
return 200, {}, "{}" return 200, {}, "{}"
def put_event_stream(self) -> TYPE_RESPONSE: def put_event_stream(self) -> str:
application_id = self.path.split("/")[-2] application_id = self.path.split("/")[-2]
params = json.loads(self.body) params = json.loads(self.body)
stream_arn = params.get("DestinationStreamArn") stream_arn = params.get("DestinationStreamArn")
@ -129,22 +99,22 @@ class PinpointResponse(BaseResponse):
) )
resp = event_stream.to_json() resp = event_stream.to_json()
resp["ApplicationId"] = application_id 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] application_id = self.path.split("/")[-2]
event_stream = self.pinpoint_backend.get_event_stream( event_stream = self.pinpoint_backend.get_event_stream(
application_id=application_id application_id=application_id
) )
resp = event_stream.to_json() resp = event_stream.to_json()
resp["ApplicationId"] = application_id 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] application_id = self.path.split("/")[-2]
event_stream = self.pinpoint_backend.delete_event_stream( event_stream = self.pinpoint_backend.delete_event_stream(
application_id=application_id application_id=application_id
) )
resp = event_stream.to_json() resp = event_stream.to_json()
resp["ApplicationId"] = application_id resp["ApplicationId"] = application_id
return 200, {}, json.dumps(resp) return json.dumps(resp)

14
moto/pinpoint/urls.py
View File

@ -7,16 +7,10 @@ url_bases = [
url_paths = { url_paths = {
"{0}/v1/apps$": PinpointResponse.method_dispatch(PinpointResponse.apps), "{0}/v1/apps$": PinpointResponse.dispatch,
"{0}/v1/apps/(?P<app_id>[^/]+)$": PinpointResponse.method_dispatch( "{0}/v1/apps/(?P<app_id>[^/]+)$": PinpointResponse.dispatch,
PinpointResponse.app "{0}/v1/apps/(?P<app_id>[^/]+)/eventstream": PinpointResponse.dispatch,
), "{0}/v1/apps/(?P<app_id>[^/]+)/settings$": PinpointResponse.dispatch,
"{0}/v1/apps/(?P<app_id>[^/]+)/eventstream": PinpointResponse.method_dispatch(
PinpointResponse.eventstream
),
"{0}/v1/apps/(?P<app_id>[^/]+)/settings$": PinpointResponse.method_dispatch(
PinpointResponse.app_settings
),
"{0}/v1/tags/(?P<app_arn>[^/]+)$": PinpointResponse.method_dispatch( "{0}/v1/tags/(?P<app_arn>[^/]+)$": PinpointResponse.method_dispatch(
PinpointResponse.tags PinpointResponse.tags
), ),