Conversation
| return self._to_raw_locations(elements) | ||
|
|
||
| def _build_query(self) -> str: | ||
| s, w, n, e = self.bbox |
There was a problem hiding this comment.
Since most queriers will support boundaries, it might be good to define a class to represent the boundaries. It'll help prevent mix-ups if another querier gets the order wrong.
| def _run(self, query: str, attempts: int = 3) -> list[dict]: | ||
| for attempt in range(attempts): | ||
| response = httpx.post( | ||
| OVERPASS_URL, data={"data": query}, |
There was a problem hiding this comment.
It's good practice for each arg to live on it's own line when they're broken up like this
There was a problem hiding this comment.
Good job making sure this included tests. It's very easy to exclude them when no one is forcing you to.
There was a problem hiding this comment.
Looks good overall, nice work!
The Only changes I think are needed include:
- separating the unit tests from the smoke tests since I believe that's what we agreed on during the last hackathon
- prepending the ticket number to the commit message i.e.
#16 Add OpenStreetMap querier with unit and smoke tests
Everything else is optional.
| @@ -0,0 +1,175 @@ | |||
| """Tests for the OpenStreetMap querier. | |||
|
|
|||
| Unit tests run offline and cover query building and RawLocation mapping. | |||
There was a problem hiding this comment.
nit(pick) - a unit test that makes a web request imo isn't a unit test.
Probably fine to omit that:
Unit tests
run offline andcover query building and RawLocation mapping.
|
|
||
| def test_build_query_exact_output(): | ||
| q = OpenStreetMapQuerier(tag_filters=['["shop"="tailor"]'], bbox=(1.0, 2.0, 3.0, 4.0), timeout_s=25) | ||
| assert q._build_query() == '[out:json][timeout:25];(nwr["shop"="tailor"](1.0,2.0,3.0,4.0););out center;' |
There was a problem hiding this comment.
I'm not normally a fan of writing unit tests for private methods (or having public methods that are only externally called by unit tests).
However, since the alternative to is use mocking & over-complicate things, it's fine as is.
This is more for future reference.
| def test_payload_is_preserved_unmodified(querier): | ||
| result = querier._to_raw_locations([WAY_WITH_CENTER]) | ||
| # Interpretation belongs to the normalizer, so the raw element passes through intact. | ||
| assert result[0].payload == WAY_WITH_CENTER |
There was a problem hiding this comment.
whenever you're checking an index, it's good to assert the result has the expected length (similar to other tests)
| RETRYABLE = {429, 502, 503, 504} | ||
|
|
||
|
|
||
| # based on coordinates set in repair-00.json in /data-explorations |
There was a problem hiding this comment.
Probably best to not reference files outside of the ETL pipeline.
Can lead to confusion if they end up being removed.
| headers={"User-Agent": USER_AGENT}, | ||
| ) | ||
| if response.status_code == 200: | ||
| return response.json()["elements"] |
There was a problem hiding this comment.
nit - it's sometimes nice to separate method calls from indexing:
body = response.json()
return body["elements"]|
|
||
| @pytest.mark.network | ||
| def test_smoke_fetch_against_live_overpass(querier): | ||
| results = querier.fetch() |
There was a problem hiding this comment.
This may throw. Might be good to catch the exception & evaluate why.
| assert all(i.split("/")[0] in {"node", "way", "relation"} for i in ids) | ||
|
|
||
| # Every element must be placeable: nodes carry lat/lon, ways/relations carry center. | ||
| for raw in results: |
There was a problem hiding this comment.
Feel like were iterating on results in different places we we could iterate once & validate in a separate method:
for raw in results:
validate_result(raw)
# or even break it up into a couple methods
validate_id(raw.data_source_id)
validate_tags(raw.payload)| assert ("lat" in p and "lon" in p) or "center" in p, f"no coordinates: {raw.data_source_id}" | ||
|
|
||
| named = [r for r in results if r.payload.get("tags", {}).get("name")] | ||
| print(f" {len(named)} named, {len(results) - len(named)} unnamed") |
There was a problem hiding this comment.
does pytest have some sort of logger? Would give some control over where & when these message would get printed. Would probably label these as debug logs.
| if response.status_code == 200: | ||
| return response.json()["elements"] | ||
| if response.status_code in RETRYABLE and attempt < attempts - 1: | ||
| time.sleep(2 ** attempt * 5) # 5s, 10s |
There was a problem hiding this comment.
Good that we're waiting longer each attempt. Most APIs expect this to some degree.
| if response.status_code in RETRYABLE and attempt < attempts - 1: | ||
| time.sleep(2 ** attempt * 5) # 5s, 10s | ||
| continue | ||
| raise RuntimeError(f"Overpass returned {response.status_code}: {response.text[:500]}") |
There was a problem hiding this comment.
Couldn't hurt to include the attempt # & an explanation if there is one. Looking at the httpx documentation, maybe reason_phrase?
2 participants
Fetches clothing-related locations from the Overpass API:
Dedupes by "{type}/{id}" so each source list has no duplicate
entries, per MergeProcessor.match()'s assumption
Sets a User-Agent (Overpass returns 406 without one) and retries after 429/502/503/504 error responses
Adds unit tests and a smoke test against live Overpass