Skip to content

Commit

Permalink
Techdebt: Simplify PinPoint routing (#7191)
Browse files Browse the repository at this point in the history
  • Loading branch information
@bblommers
bblommers authored Jan 6, 2024
1 parent 57bc607 commit e7c86e5
Show file tree
Hide file tree
Showing 2 changed files with 20 additions and 56 deletions.
62 changes: 16 additions & 46 deletions moto/pinpoint/responses.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,36 +20,6 @@ def pinpoint_backend(self) -> PinpointBackend:
"""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":
Expand All @@ -66,39 +36,39 @@ def create_app(self) -> TYPE_RESPONSE:
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(
application_id=application_id, settings=settings
)
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]
Expand All @@ -119,7 +89,7 @@ def untag_resource(self) -> TYPE_RESPONSE:
)
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")
Expand All @@ -129,22 +99,22 @@ def put_event_stream(self) -> TYPE_RESPONSE:
)
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)
14 changes: 4 additions & 10 deletions moto/pinpoint/urls.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,16 +7,10 @@


url_paths = {
"{0}/v1/apps$": PinpointResponse.method_dispatch(PinpointResponse.apps),
"{0}/v1/apps/(?P<app_id>[^/]+)$": PinpointResponse.method_dispatch(
PinpointResponse.app
),
"{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/apps$": PinpointResponse.dispatch,
"{0}/v1/apps/(?P<app_id>[^/]+)$": PinpointResponse.dispatch,
"{0}/v1/apps/(?P<app_id>[^/]+)/eventstream": PinpointResponse.dispatch,
"{0}/v1/apps/(?P<app_id>[^/]+)/settings$": PinpointResponse.dispatch,
"{0}/v1/tags/(?P<app_arn>[^/]+)$": PinpointResponse.method_dispatch(
PinpointResponse.tags
),
Expand Down

0 comments on commit e7c86e5

Please sign in to comment.