w1cdn/spothole
1
0
Fork 0
mirror of https://git.ianrenton.com/ian/spothole.git synced 2026-09-28 18:22:05 +00:00
Code Issues 13 Packages Projects Releases Wiki Activity

Multi select upstream providers to send spots to #95

Browse Source
This commit is contained in:
Ian Renton
2026-09-27 14:37:48 +01:00
parent 0cfe92db01
commit e6f0f737d0
13 changed files with 178 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
webserver/handlers/api/addspot.py
+57 -43
View File
@@ -1,3 +1,4 @@
import asyncio
import logging
import re
import threading
@@ -78,9 +79,9 @@ class APISpotHandler(tornado.web.RequestHandler):
handling = json_body.get("handling", {})
# Extract individual parameters that say how this spot should be handled by the server
submit_upstream = handling.get("submit_upstream", False)
upstream_provider_name = handling.get("upstream_provider", None)
upstream_credentials = handling.get("upstream_credentials", {})
upstream_provider_names = handling.get("upstream_providers", None) or []
upstream_credentials = handling.get("upstream_credentials", None) or {}
submit_upstream = len(upstream_provider_names) > 0
captcha_token = handling.get("captcha_token", None)
# Verify CAPTCHA if required
@@ -166,13 +167,18 @@ class APISpotHandler(tornado.web.RequestHandler):
self.set_header("Content-Type", "application/json")
return
# Validate upstream submission requirements
if submit_upstream and upstream_provider_name:
is_cluster = isinstance(self._find_provider(upstream_provider_name, spot.activities), DXCluster)
is_tiles = isinstance(self._find_provider(upstream_provider_name, spot.activities), Tiles)
# Validate upstream submission requirements for each requested provider
for upstream_provider_name in upstream_provider_names:
provider = self._find_provider(upstream_provider_name, spot.activities)
is_cluster = isinstance(provider, DXCluster)
is_tiles = isinstance(provider, Tiles)
if not spot.activity_refs and not is_tiles and not is_cluster:
self.set_status(422)
self.write(safe_json_dumps("Error - an activity reference is required to submit upstream."))
self.write(
safe_json_dumps(
f"Error - an activity reference is required to submit upstream to {upstream_provider_name}."
)
)
self.set_header("Cache-Control", "no-store")
self.set_header("Content-Type", "application/json")
return
@@ -191,46 +197,27 @@ class APISpotHandler(tornado.web.RequestHandler):
self.set_header("Content-Type", "application/json")
return
# Submit upstream if requested
upstream_warning = None
if submit_upstream and upstream_provider_name:
provider = self._find_provider(upstream_provider_name, spot.activities)
if provider:
try:
# Submit spot to the upstream provider. Run in a separate thread otherwise this blocks the whole
# web server for everyone!
await IOLoop.current().run_in_executor(None, provider.submit_spot, spot, upstream_credentials)
# Trigger a re-poll after 3 second so the spot appears quickly. (Submitting to a cluster node is
# slower than this, but we get data as a live stream from cluster anyway, so force_poll does
# nothing in that case. This is really just for the HTTP providers when we submit a spot to them)
threading.Timer(3.0, provider.force_poll).start()
except NotImplementedError as e:
upstream_warning = str(e)
except SpotSubmissionError as e:
logger.warning(f"Upstream submission to {upstream_provider_name} was not accepted: {e}")
upstream_warning = (
f"Spot was saved locally but upstream submission to {upstream_provider_name} failed: {e}"
)
except Exception:
logger.exception(f"Failed to submit spot upstream to {upstream_provider_name}")
upstream_warning = (
f"Spot was saved locally but upstream submission to {upstream_provider_name} failed."
)
elif spot.activities:
upstream_warning = f"No enabled provider named '{upstream_provider_name}' supports upstream submission for {', '.join(spot.activities)} spots."
else:
upstream_warning = f"No enabled provider named '{upstream_provider_name}' supports upstream submission for spots with no activity."
# Submit upstream to all requested providers in parallel, collecting any warnings
results = await asyncio.gather(
*(
self._submit_upstream(name, spot, upstream_credentials.get(name, {}))
for name in upstream_provider_names
)
)
upstream_warnings = [w for w in results if w]
any_upstream_succeeded = len(upstream_warnings) < len(upstream_provider_names)
# If we successfully submitted the spot upstream, don't add it direct to Spothole, otherwise it will be a
# duplicate with what immediately comes back from the API. But if we weren't asked to send it upstream, or
# we were but it failed, we should still add it to our database anyway.
if not submit_upstream or upstream_warning:
# If we successfully submitted the spot to at least one upstream provider, don't add it direct to Spothole,
# otherwise it will be a duplicate with what immediately comes back from the API. But if we weren't asked to
# send it upstream, or we were but every submission failed, we should still add it to our database anyway.
if not any_upstream_succeeded:
spot.source = "API"
spot.infer_missing()
self._spots.set(spot.id, spot)
if upstream_warning:
self.write(safe_json_dumps(f"Warning - {upstream_warning}"))
if upstream_warnings:
saved_locally_note = "" if any_upstream_succeeded else " The spot was saved to Spothole only."
self.write(safe_json_dumps(f"Warning - {' '.join(upstream_warnings)}{saved_locally_note}"))
self.set_status(201)
else:
self.write(safe_json_dumps("OK"))
@@ -245,6 +232,33 @@ class APISpotHandler(tornado.web.RequestHandler):
self.set_header("Cache-Control", "no-store")
self.set_header("Content-Type", "application/json")
async def _submit_upstream(self, upstream_provider_name, spot, credentials) -> str | None:
"""Submit a spot to the named upstream provider. Returns None on success, or a warning message on failure."""
provider = self._find_provider(upstream_provider_name, spot.activities)
if not provider:
if spot.activities:
return f"No enabled provider named '{upstream_provider_name}' supports upstream submission for {', '.join(spot.activities)} spots."
return f"No enabled provider named '{upstream_provider_name}' supports upstream submission for spots with no activity."
try:
# Submit spot to the upstream provider. Run in a separate thread otherwise this blocks the whole web server
# for everyone!
await IOLoop.current().run_in_executor(None, provider.submit_spot, spot, credentials)
# Trigger a re-poll after 3 second so the spot appears quickly. (Submitting to a cluster node is slower than
# this, but we get data as a live stream from cluster anyway, so force_poll does nothing in that case. This
# is really just for the HTTP providers when we submit a spot to them)
threading.Timer(3.0, provider.force_poll).start()
return None
except NotImplementedError as e:
return str(e)
except SpotSubmissionError as e:
logger.warning(f"Upstream submission to {upstream_provider_name} was not accepted: {e}")
return f"Upstream submission to {upstream_provider_name} failed: {e}"
except Exception:
logger.exception(f"Failed to submit spot upstream to {upstream_provider_name}")
return f"Upstream submission to {upstream_provider_name} failed."
def _find_provider(self, provider_name, activities) -> SpotProvider | None:
"""Find an enabled provider by name that can submit spots for at least one of the given activities. If there
are no activities, find one that can submit spots with no activity."""
webserver/handlers/api/compatibility/v2_compatibility.py
+17
View File
@@ -116,6 +116,8 @@ class V2APISpotHandler(V2CompatibilityWrapper, RequestCompatibilityWrapper, APIS
def translate(body):
if isinstance(body, dict) and isinstance(body.get("spot"), dict):
body["spot"] = self._translate_v2_spot(body["spot"])
if isinstance(body, dict) and isinstance(body.get("handling"), dict):
body["handling"] = self._translate_v2_handling(body["handling"])
return body
translate_json_body(self.request, translate)
@@ -141,6 +143,21 @@ class V2APISpotHandler(V2CompatibilityWrapper, RequestCompatibilityWrapper, APIS
spot_data["activity_refs"] = refs
return spot_data
@staticmethod
def _translate_v2_handling(handling):
"""Translate the "handling" object provided by a client calling the add spot method in v2 format into v3
format. v2 had a "submit_upstream" flag with a single "upstream_provider" and its credentials, whereas v3 has
a list of "upstream_providers" and a map of provider name to credentials."""
handling = dict(handling)
submit_upstream = handling.pop("submit_upstream", False)
provider_name = handling.pop("upstream_provider", None)
credentials = handling.pop("upstream_credentials", None) or {}
if submit_upstream and provider_name:
handling["upstream_providers"] = [provider_name]
handling["upstream_credentials"] = {provider_name: credentials}
return handling
def collapse_activities(obj):
"""Utility method to replace the "activities" list in a spot or alert JSON object with a single "activity" value,