diff --git a/api/app.py b/api/app.py index 3940056..f864712 100644 --- a/api/app.py +++ b/api/app.py @@ -1,3 +1,4 @@ +import os from contextlib import asynccontextmanager from pathlib import Path @@ -20,7 +21,8 @@ @asynccontextmanager async def lifespan(app: Litestar): await open_pool() - await ensure_schema() + if os.getenv("OSMSG_SKIP_SCHEMA_ENSURE", "").lower() not in {"1", "true", "yes"}: + await ensure_schema() try: yield finally: @@ -49,7 +51,11 @@ async def health() -> HealthResponse: app = Litestar( route_handlers=[home, health, v1_router], lifespan=[lifespan], - cors_config=CORSConfig(allow_origins=["*"]), + cors_config=CORSConfig( + allow_origins=["*"], + allow_methods=["GET", "OPTIONS"], + allow_headers=["*"], + ), openapi_config=OpenAPIConfig( title="OSMSG API", version="1.0.0", diff --git a/api/pagination.py b/api/pagination.py new file mode 100644 index 0000000..2785178 --- /dev/null +++ b/api/pagination.py @@ -0,0 +1,22 @@ +from typing import TypeVar + +from litestar.pagination import OffsetPagination +from pydantic import BaseModel, Field + +RowT = TypeVar("RowT") + + +class PaginationParams(BaseModel): + limit: int = Field(default=100, ge=1, le=1000) + offset: int = Field(default=0, ge=0) + + @property + def query_limit(self) -> int: + return self.limit + 1 + + +def paginate_items(rows: list[RowT], params: PaginationParams) -> OffsetPagination[RowT]: + has_next = len(rows) > params.limit + items = rows[: params.limit] + total = params.offset + len(items) + int(has_next) + return OffsetPagination(items=items, limit=params.limit, offset=params.offset, total=total) diff --git a/api/pg_schema.py b/api/pg_schema.py index e22aec4..c1826c9 100644 --- a/api/pg_schema.py +++ b/api/pg_schema.py @@ -1,5 +1,4 @@ PG_SCHEMA = """ -CREATE EXTENSION IF NOT EXISTS postgis; CREATE TABLE IF NOT EXISTS users ( uid BIGINT PRIMARY KEY, username TEXT NOT NULL @@ -10,10 +9,17 @@ created_at TIMESTAMPTZ, hashtags TEXT[], editor TEXT, - geom GEOMETRY(POLYGON) + min_lon DOUBLE PRECISION, + min_lat DOUBLE PRECISION, + max_lon DOUBLE PRECISION, + max_lat DOUBLE PRECISION +); +CREATE INDEX IF NOT EXISTS idx_changesets_created_at ON changesets USING BTREE (created_at); +CREATE INDEX IF NOT EXISTS idx_changesets_hashtags ON changesets USING GIN (hashtags); +CREATE INDEX IF NOT EXISTS idx_changesets_editor ON changesets USING BTREE (editor); +CREATE INDEX IF NOT EXISTS idx_changesets_bbox ON changesets USING GIST ( + box(point(min_lon, min_lat), point(max_lon, max_lat)) ); -CREATE INDEX IF NOT EXISTS idx_changesets_created_at ON changesets(created_at); -CREATE INDEX IF NOT EXISTS idx_changesets_geom ON changesets USING GIST (geom); CREATE TABLE IF NOT EXISTS changeset_stats ( changeset_id BIGINT NOT NULL REFERENCES changesets(changeset_id), seq_id BIGINT NOT NULL, @@ -32,8 +38,8 @@ tag_stats JSONB, PRIMARY KEY (seq_id, changeset_id) ); -CREATE INDEX IF NOT EXISTS idx_changeset_stats_uid ON changeset_stats(uid); -CREATE INDEX IF NOT EXISTS idx_changeset_stats_changeset_id ON changeset_stats(changeset_id); +CREATE INDEX IF NOT EXISTS idx_changeset_stats_uid ON changeset_stats USING BTREE (uid); +CREATE INDEX IF NOT EXISTS idx_changeset_stats_changeset_id ON changeset_stats USING BTREE (changeset_id); CREATE TABLE IF NOT EXISTS state ( source_url TEXT PRIMARY KEY, last_seq BIGINT NOT NULL, diff --git a/api/queries.py b/api/queries.py index b894258..5f0f9ea 100644 --- a/api/queries.py +++ b/api/queries.py @@ -1,11 +1,129 @@ +import json from datetime import datetime from typing import Any from .db import get_pool +GEOMETRY_MODE_GEOM = "geom" +GEOMETRY_MODE_BBOX = "bbox" -def _user_stats_sql(*, filter_dates: bool, filter_hashtags: bool, include_tags: bool) -> str: + +def _map_changes_expr(alias: str = "st") -> str: + return f""" + {alias}.nodes_created + {alias}.nodes_modified + {alias}.nodes_deleted + + {alias}.ways_created + {alias}.ways_modified + {alias}.ways_deleted + + {alias}.rels_created + {alias}.rels_modified + {alias}.rels_deleted + """ + + +_TAG_VALUE_CTES = """, + tag_agg AS ( + SELECT + st.uid, + tk.key AS tag_key, + tv.key AS tag_val, + SUM(COALESCE((tv.value->>'c')::bigint, 0)) AS total_c, + SUM(COALESCE((tv.value->>'m')::bigint, 0)) AS total_m, + SUM((tv.value->>'len')::double precision) AS total_len + FROM stats_scope st + JOIN LATERAL jsonb_each(st.tag_stats) tk ON st.tag_stats IS NOT NULL + JOIN LATERAL jsonb_each(tk.value) tv ON true + GROUP BY st.uid, tk.key, tv.key + ), + tag_per_key AS ( + SELECT + uid, + tag_key, + jsonb_object_agg( + tag_val, + CASE WHEN total_len IS NOT NULL + THEN jsonb_build_object('c', total_c, 'm', total_m, 'len', total_len) + ELSE jsonb_build_object('c', total_c, 'm', total_m) + END + ) AS tag_vals + FROM tag_agg + GROUP BY uid, tag_key + ), + tag_per_user AS ( + SELECT uid, jsonb_object_agg(tag_key, tag_vals) AS tag_stats + FROM tag_per_key + GROUP BY uid + )""" + +_TAG_KEY_CTES = """, + tag_key_agg AS ( + SELECT + st.uid, + tk.key AS tag_key, + SUM(COALESCE(key_stats.total_c, 0)) AS total_c, + SUM(COALESCE(key_stats.total_m, 0)) AS total_m, + SUM(key_stats.total_len) AS total_len + FROM stats_scope st + JOIN LATERAL jsonb_each(st.tag_stats) tk ON st.tag_stats IS NOT NULL + JOIN LATERAL ( + SELECT + SUM(COALESCE((tv.value->>'c')::bigint, 0)) AS total_c, + SUM(COALESCE((tv.value->>'m')::bigint, 0)) AS total_m, + SUM((tv.value->>'len')::double precision) AS total_len + FROM jsonb_each(tk.value) tv + ) key_stats ON true + GROUP BY st.uid, tk.key + ), + tag_per_user AS ( + SELECT + uid, + jsonb_object_agg( + tag_key, + jsonb_build_object('c', total_c, 'm', total_m, 'len', total_len) + ) AS tag_stats + FROM tag_key_agg + GROUP BY uid + )""" + +_HASHTAG_CTE = """, + user_hashtags AS ( + SELECT + st.uid, + ARRAY_AGG(DISTINCT ht.hashtag ORDER BY ht.hashtag) AS hashtags + FROM stats_scope st + JOIN changesets cs ON cs.changeset_id = st.changeset_id + CROSS JOIN LATERAL UNNEST(cs.hashtags) AS ht(hashtag) + WHERE cs.hashtags IS NOT NULL + GROUP BY st.uid + )""" + + +def _column(alias: str, name: str) -> str: + return f"{alias}.{name}" if alias else name + + +def _bbox_centroid_geojson_sql(alias: str = "cs") -> str: + min_lon = _column(alias, "min_lon") + min_lat = _column(alias, "min_lat") + max_lon = _column(alias, "max_lon") + max_lat = _column(alias, "max_lat") + return f""" + CASE + WHEN {min_lon} IS NULL OR {min_lat} IS NULL OR {max_lon} IS NULL OR {max_lat} IS NULL THEN NULL + ELSE jsonb_build_object( + 'type', 'Point', + 'coordinates', jsonb_build_array( + ({min_lon} + {max_lon}) / 2.0, + ({min_lat} + {max_lat}) / 2.0 + ) + )::TEXT + END + """ + + +def _user_stats_sql( + *, + filter_dates: bool, + filter_hashtags: bool, + tag_mode: str, +) -> str: n = 1 + include_tags = tag_mode != "none" changeset_filters: list[str] = [] if filter_dates: @@ -81,8 +199,39 @@ def _user_stats_sql(*, filter_dates: bool, filter_hashtags: bool, include_tags: GROUP BY st.uid )""" - tag_ctes = ( - """, + if tag_mode == "keys": + tag_ctes = """, + tag_key_agg AS ( + SELECT + st.uid, + tk.key AS tag_key, + SUM(COALESCE(key_stats.total_c, 0)) AS total_c, + SUM(COALESCE(key_stats.total_m, 0)) AS total_m, + SUM(key_stats.total_len) AS total_len + FROM stats_scope st + JOIN ranked r ON r.uid = st.uid + JOIN LATERAL jsonb_each(st.tag_stats) tk ON st.tag_stats IS NOT NULL + JOIN LATERAL ( + SELECT + SUM(COALESCE((tv.value->>'c')::bigint, 0)) AS total_c, + SUM(COALESCE((tv.value->>'m')::bigint, 0)) AS total_m, + SUM((tv.value->>'len')::double precision) AS total_len + FROM jsonb_each(tk.value) tv + ) key_stats ON true + GROUP BY st.uid, tk.key + ), + tag_per_user AS ( + SELECT + uid, + jsonb_object_agg( + tag_key, + jsonb_build_object('c', total_c, 'm', total_m, 'len', total_len) + ) AS tag_stats + FROM tag_key_agg + GROUP BY uid + )""" + elif tag_mode == "all": + tag_ctes = """, tag_agg AS ( SELECT st.uid, @@ -116,11 +265,10 @@ def _user_stats_sql(*, filter_dates: bool, filter_hashtags: bool, include_tags: FROM tag_per_key GROUP BY uid )""" - if include_tags - else "" - ) + else: + tag_ctes = "" - tag_select = "tpu.tag_stats" if include_tags else "NULL::jsonb AS tag_stats" + tag_select = "COALESCE(tpu.tag_stats, '{}'::jsonb) AS tag_stats" if include_tags else "NULL::jsonb AS tag_stats" tag_join = "LEFT JOIN tag_per_user tpu ON tpu.uid = r.uid" if include_tags else "" return f""" @@ -145,6 +293,250 @@ def _user_stats_sql(*, filter_dates: bool, filter_hashtags: bool, include_tags: """ +def _changeset_filters_sql( + *, + filter_dates: bool, + filter_hashtags: bool = False, +) -> tuple[str, int]: + n = 1 + filters: list[str] = [] + if filter_dates: + filters.append(f"cs.created_at >= ${n}") + n += 1 + filters.append(f"cs.created_at < ${n}") + n += 1 + if filter_hashtags: + filters.append(f"cs.hashtags && ${n}::TEXT[]") + n += 1 + where_sql = f"WHERE {' AND '.join(filters)}" if filters else "" + return where_sql, n + + +def _hashtag_stats_sql( + *, + filter_dates: bool, + filter_hashtags: bool, +) -> str: + where_sql, n = _changeset_filters_sql( + filter_dates=filter_dates, + filter_hashtags=filter_hashtags, + ) + hashtag_param = f"${3 if filter_dates else 1}" if filter_hashtags else None + hashtag_value_filter = f"AND ht.hashtag = ANY({hashtag_param}::TEXT[])" if hashtag_param else "" + limit_param = f"${n}" + offset_param = f"${n + 1}" + map_changes = _map_changes_expr() + return f""" + WITH hashtag_scope AS ( + SELECT + ht.hashtag, + st.uid, + st.changeset_id, + ({map_changes}) AS map_changes + FROM changesets cs + JOIN changeset_stats st ON st.changeset_id = cs.changeset_id + CROSS JOIN LATERAL UNNEST(cs.hashtags) AS ht(hashtag) + {where_sql} + {hashtag_value_filter} + ), + hashtag_totals AS ( + SELECT + hashtag, + COUNT(DISTINCT changeset_id) AS changesets, + COUNT(DISTINCT uid) AS users, + COALESCE(SUM(map_changes), 0) AS map_changes + FROM hashtag_scope + GROUP BY hashtag + ) + SELECT + hashtag, + changesets, + users, + map_changes, + ROW_NUMBER() OVER (ORDER BY map_changes DESC, hashtag ASC) AS rank + FROM hashtag_totals + ORDER BY map_changes DESC, hashtag ASC + LIMIT {limit_param} OFFSET {offset_param} + """ + + +def _hashtag_trends_sql(*, filter_hashtags: bool) -> str: + where_sql, n = _changeset_filters_sql( + filter_dates=True, + filter_hashtags=filter_hashtags, + ) + hashtag_param = "$3" if filter_hashtags else None + hashtag_value_filter = f"AND ht.hashtag = ANY({hashtag_param}::TEXT[])" if hashtag_param else "" + interval_param = f"${n}" + limit_param = f"${n + 1}" + offset_param = f"${n + 2}" + map_changes = _map_changes_expr() + return f""" + SELECT + DATE_TRUNC({interval_param}, cs.created_at) AS period_start, + ht.hashtag, + COUNT(DISTINCT st.changeset_id) AS changesets, + COUNT(DISTINCT st.uid) AS users, + COALESCE(SUM({map_changes}), 0) AS map_changes + FROM changesets cs + JOIN changeset_stats st ON st.changeset_id = cs.changeset_id + CROSS JOIN LATERAL UNNEST(cs.hashtags) AS ht(hashtag) + {where_sql} + {hashtag_value_filter} + GROUP BY period_start, ht.hashtag + ORDER BY period_start ASC, map_changes DESC, ht.hashtag ASC + LIMIT {limit_param} OFFSET {offset_param} + """ + + +def _editor_stats_sql(*, filter_dates: bool, include_version: bool) -> str: + where_sql, n = _changeset_filters_sql(filter_dates=filter_dates) + limit_param = f"${n}" + offset_param = f"${n + 1}" + map_changes = _map_changes_expr() + editor_expr = ( + "COALESCE(NULLIF(cs.editor, ''), 'unknown')" + if include_version + else "COALESCE(NULLIF(REGEXP_REPLACE(cs.editor, E'([ /]v?[0-9].*)$', '', 'i'), ''), 'unknown')" + ) + return f""" + WITH editor_scope AS ( + SELECT + {editor_expr} AS editor, + st.uid, + st.changeset_id, + ({map_changes}) AS map_changes + FROM changesets cs + JOIN changeset_stats st ON st.changeset_id = cs.changeset_id + {where_sql} + ), + editor_totals AS ( + SELECT + editor, + COUNT(DISTINCT changeset_id) AS changesets, + COUNT(DISTINCT uid) AS users, + COALESCE(SUM(map_changes), 0) AS map_changes + FROM editor_scope + GROUP BY editor + ) + SELECT + editor, + changesets, + users, + map_changes, + ROW_NUMBER() OVER (ORDER BY map_changes DESC, editor ASC) AS rank + FROM editor_totals + ORDER BY map_changes DESC, editor ASC + LIMIT {limit_param} OFFSET {offset_param} + """ + + +def _map_geometry_sql(mode: str | None) -> tuple[str, str]: + if mode == GEOMETRY_MODE_GEOM: + return ( + "ST_AsGeoJSON(ST_Centroid(cs.geom))::TEXT AS geometry", + "cs.geom", + ) + if mode == GEOMETRY_MODE_BBOX: + return ( + f"{_bbox_centroid_geojson_sql()} AS geometry", + "cs.min_lon, cs.min_lat, cs.max_lon, cs.max_lat", + ) + return "NULL::TEXT AS geometry", "" + + +def _map_geometry_filter_sql(mode: str | None) -> str: + if mode == GEOMETRY_MODE_GEOM: + return "cs.geom IS NOT NULL" + if mode == GEOMETRY_MODE_BBOX: + return " AND ".join( + [ + "cs.min_lon IS NOT NULL", + "cs.min_lat IS NOT NULL", + "cs.max_lon IS NOT NULL", + "cs.max_lat IS NOT NULL", + ] + ) + return "FALSE" + + +def _map_changes_sql( + *, + filter_dates: bool, + filter_hashtags: bool, + geometry_mode: str | None, +) -> str: + where_sql, n = _changeset_filters_sql( + filter_dates=filter_dates, + filter_hashtags=filter_hashtags, + ) + limit_param = f"${n}" + offset_param = f"${n + 1}" + map_changes = _map_changes_expr() + geometry_select, geometry_group = _map_geometry_sql(geometry_mode) + geometry_group_sql = f", {geometry_group}" if geometry_group else "" + geometry_filter = _map_geometry_filter_sql(geometry_mode) + located_where_sql = f"{where_sql} AND {geometry_filter}" if where_sql else f"WHERE {geometry_filter}" + return f""" + WITH filtered_changesets AS ( + SELECT cs.* + FROM changesets cs + {located_where_sql} + ORDER BY cs.created_at DESC NULLS LAST, cs.changeset_id DESC + LIMIT {limit_param} OFFSET {offset_param} + ) + SELECT + cs.changeset_id, + cs.uid, + u.username AS name, + cs.created_at, + COALESCE(cs.hashtags, ARRAY[]::TEXT[]) AS hashtags, + COALESCE(NULLIF(cs.editor, ''), 'unknown') AS editor, + COALESCE(SUM({map_changes}), 0) AS map_changes, + COALESCE(SUM(st.nodes_created), 0) AS nodes_create, + COALESCE(SUM(st.nodes_modified), 0) AS nodes_modify, + COALESCE(SUM(st.nodes_deleted), 0) AS nodes_delete, + COALESCE(SUM(st.ways_created), 0) AS ways_create, + COALESCE(SUM(st.ways_modified), 0) AS ways_modify, + COALESCE(SUM(st.ways_deleted), 0) AS ways_delete, + COALESCE(SUM(st.rels_created), 0) AS rels_create, + COALESCE(SUM(st.rels_modified), 0) AS rels_modify, + COALESCE(SUM(st.rels_deleted), 0) AS rels_delete, + {geometry_select} + FROM filtered_changesets cs + JOIN users u ON u.uid = cs.uid + LEFT JOIN changeset_stats st ON st.changeset_id = cs.changeset_id + GROUP BY cs.changeset_id, cs.uid, u.username, cs.created_at, cs.hashtags, cs.editor{geometry_group_sql} + ORDER BY cs.created_at DESC NULLS LAST, cs.changeset_id DESC + """ + + +async def _changesets_columns() -> set[str]: + async with get_pool().acquire() as conn: + rows = await conn.fetch( + """ + SELECT column_name + FROM information_schema.columns + WHERE table_name = 'changesets' + """ + ) + return {row["column_name"] for row in rows} + + +async def _postgis_available() -> bool: + async with get_pool().acquire() as conn: + return bool(await conn.fetchval("SELECT COUNT(*) FROM pg_extension WHERE extname = 'postgis'")) + + +async def _changesets_geometry_mode() -> str | None: + columns = await _changesets_columns() + if "geom" in columns and await _postgis_available(): + return GEOMETRY_MODE_GEOM + if {"min_lon", "min_lat", "max_lon", "max_lat"}.issubset(columns): + return GEOMETRY_MODE_BBOX + return None + + async def fetch_state() -> dict[str, Any] | None: # last_ts/last_seq come from the worst-lagging source (slowest source bounds real freshness); # updated_at is the most recent heartbeat across all sources (any tick proves the worker is alive). @@ -167,13 +559,17 @@ async def fetch_user_stats( start: datetime | None = None, end: datetime | None = None, hashtag: list[str] | None = None, - tags: bool = True, + tag_mode: str = "keys", limit: int = 100, offset: int = 0, ) -> list[dict[str, Any]]: filter_dates = start is not None and end is not None filter_hashtags = bool(hashtag) - sql = _user_stats_sql(filter_dates=filter_dates, filter_hashtags=filter_hashtags, include_tags=tags) + sql = _user_stats_sql( + filter_dates=filter_dates, + filter_hashtags=filter_hashtags, + tag_mode=tag_mode, + ) params: list[Any] = [] if filter_dates: params.extend([start, end]) @@ -184,3 +580,110 @@ async def fetch_user_stats( async with get_pool().acquire() as conn: rows = await conn.fetch(sql, *params) return [dict(row) for row in rows] + + +async def fetch_hashtag_stats( + *, + start: datetime | None = None, + end: datetime | None = None, + hashtag: list[str] | None = None, + limit: int = 100, + offset: int = 0, +) -> list[dict[str, Any]]: + filter_dates = start is not None and end is not None + filter_hashtags = bool(hashtag) + sql = _hashtag_stats_sql( + filter_dates=filter_dates, + filter_hashtags=filter_hashtags, + ) + params: list[Any] = [] + if filter_dates: + params.extend([start, end]) + if filter_hashtags: + params.append(hashtag) + params.extend([limit, offset]) + + async with get_pool().acquire() as conn: + rows = await conn.fetch(sql, *params) + return [dict(row) for row in rows] + + +async def fetch_hashtag_trends( + *, + start: datetime, + end: datetime, + interval: str, + hashtag: list[str] | None = None, + limit: int = 1000, + offset: int = 0, +) -> list[dict[str, Any]]: + filter_hashtags = bool(hashtag) + sql = _hashtag_trends_sql(filter_hashtags=filter_hashtags) + params: list[Any] = [start, end] + if filter_hashtags: + params.append(hashtag) + params.extend([interval, limit, offset]) + + async with get_pool().acquire() as conn: + rows = await conn.fetch(sql, *params) + return [dict(row) for row in rows] + + +async def fetch_editor_stats( + *, + start: datetime | None = None, + end: datetime | None = None, + include_version: bool = False, + limit: int = 100, + offset: int = 0, +) -> list[dict[str, Any]]: + filter_dates = start is not None and end is not None + sql = _editor_stats_sql(filter_dates=filter_dates, include_version=include_version) + params: list[Any] = [] + if filter_dates: + params.extend([start, end]) + params.extend([limit, offset]) + + async with get_pool().acquire() as conn: + rows = await conn.fetch(sql, *params) + return [dict(row) for row in rows] + + +async def fetch_map_changes( + *, + start: datetime | None = None, + end: datetime | None = None, + hashtag: list[str] | None = None, + limit: int = 500, + offset: int = 0, +) -> dict[str, Any]: + filter_dates = start is not None and end is not None + filter_hashtags = bool(hashtag) + geometry_mode = await _changesets_geometry_mode() + sql = _map_changes_sql( + filter_dates=filter_dates, + filter_hashtags=filter_hashtags, + geometry_mode=geometry_mode, + ) + params: list[Any] = [] + if filter_dates: + params.extend([start, end]) + if filter_hashtags: + params.append(hashtag) + params.extend([limit, offset]) + + async with get_pool().acquire() as conn: + rows = await conn.fetch(sql, *params) + + features = [] + for row in rows: + data = dict(row) + geometry = data.pop("geometry") + features.append( + { + "type": "Feature", + "geometry": json.loads(geometry) if geometry else None, + "properties": data, + } + ) + return {"type": "FeatureCollection", "features": features} diff --git a/api/routers/v1.py b/api/routers/v1.py index 268944d..8470633 100644 --- a/api/routers/v1.py +++ b/api/routers/v1.py @@ -1,12 +1,25 @@ -from datetime import UTC, datetime -from typing import Annotated +from datetime import UTC, datetime, timedelta +from typing import Annotated, Literal, cast from litestar import Controller, Router, get from litestar.exceptions import HTTPException from litestar.params import Parameter -from ..queries import fetch_user_stats -from ..schemas import UserStat, UserStatsResponse +from ..pagination import PaginationParams, paginate_items +from ..queries import fetch_editor_stats, fetch_hashtag_stats, fetch_hashtag_trends, fetch_map_changes, fetch_user_stats +from ..schemas import ( + EditorStat, + EditorStatsResponse, + HashtagStat, + HashtagStatsResponse, + HashtagTrend, + MapFeatureCollection, + UserStat, + UserStatsResponse, +) + +TREND_INTERVALS = {"day", "week", "month"} +TAG_MODES = {"keys", "all"} def normalize_hashtags(hashtag: list[str] | None) -> list[str] | None: @@ -27,6 +40,22 @@ def normalize_hashtags(hashtag: list[str] | None) -> list[str] | None: return normalized or None +def resolve_optional_window(start: datetime | None, end: datetime | None) -> tuple[datetime | None, datetime | None]: + start = start or (datetime.min.replace(tzinfo=UTC) if end else None) + end = end or (datetime.now(tz=UTC) if start else None) + if start and end and start >= end: + raise HTTPException(status_code=400, detail="start must be before end") + return start, end + + +def resolve_required_window(start: datetime | None, end: datetime | None) -> tuple[datetime, datetime]: + end = end or datetime.now(tz=UTC) + start = start or (end - timedelta(days=30)) + if start >= end: + raise HTTPException(status_code=400, detail="start must be before end") + return start, end + + class StatsController(Controller): path = "/stats" @@ -44,34 +73,182 @@ async def get_user_stats( list[str] | None, Parameter(description="Filter to changesets carrying any of these hashtags. Repeatable.") ] = None, tags: Annotated[bool, Parameter(description="Include per-user tag_stats breakdown in the response.")] = True, + tag_mode: Annotated[ + str, + Parameter(description="Tag aggregation: keys (default compact totals) or all (key/value details)."), + ] = "keys", limit: Annotated[int, Parameter(ge=1, le=1000, description="Page size (1 to 1000).")] = 100, offset: Annotated[int, Parameter(ge=0, description="Page offset.")] = 0, ) -> UserStatsResponse: - start = start or (datetime.min.replace(tzinfo=UTC) if end else None) - end = end or (datetime.now(tz=UTC) if start else None) - if start and end and start >= end: - raise HTTPException(status_code=400, detail="start must be before end") - + start, end = resolve_optional_window(start, end) normalized_hashtag = normalize_hashtags(hashtag) + if tag_mode not in TAG_MODES: + raise HTTPException(status_code=400, detail="tag_mode must be one of: keys, all") + resolved_tag_mode = "none" if not tags else cast(Literal["keys", "all"], tag_mode) + page_params = PaginationParams(limit=limit, offset=offset) rows = await fetch_user_stats( start=start, end=end, hashtag=normalized_hashtag, - tags=tags, + tag_mode=resolved_tag_mode, + limit=page_params.query_limit, + offset=page_params.offset, + ) + page = paginate_items([UserStat(**row) for row in rows], page_params) + return UserStatsResponse( + count=len(page.items), + start=start, + end=end, + hashtag=normalized_hashtag, + tags=resolved_tag_mode != "none", + tag_mode=resolved_tag_mode, limit=limit, offset=offset, + pagination=page, + users=page.items, ) - users = [UserStat(**row) for row in rows] - return UserStatsResponse( - count=len(users), + + +class HashtagStatsController(Controller): + path = "/hashtag-stats" + + @get() + async def get_hashtag_stats( + self, + start: Annotated[ + datetime | None, + Parameter(description="Inclusive UTC lower bound (ISO 8601). Defaults to 30 days before end."), + ] = None, + end: Annotated[ + datetime | None, + Parameter(description="Exclusive UTC upper bound (ISO 8601). Defaults to now."), + ] = None, + hashtag: Annotated[ + list[str] | None, Parameter(description="Optional hashtags to limit the leaderboard to. Repeatable.") + ] = None, + interval: Annotated[str, Parameter(description="Trend bucket: day, week, or month.")] = "day", + limit: Annotated[int, Parameter(ge=1, le=1000, description="Page size (1-1000).")] = 100, + offset: Annotated[int, Parameter(ge=0, description="Page offset.")] = 0, + ) -> HashtagStatsResponse: + if interval not in TREND_INTERVALS: + raise HTTPException(status_code=400, detail="interval must be one of: day, week, month") + + start, end = resolve_required_window(start, end) + normalized_hashtag = normalize_hashtags(hashtag) + page_params = PaginationParams(limit=limit, offset=offset) + hashtag_rows = await fetch_hashtag_stats( + start=start, + end=end, + hashtag=normalized_hashtag, + limit=page_params.query_limit, + offset=page_params.offset, + ) + trend_rows = await fetch_hashtag_trends( + start=start, + end=end, + interval=interval, + hashtag=normalized_hashtag, + limit=page_params.query_limit, + offset=page_params.offset, + ) + hashtag_page = paginate_items([HashtagStat(**row) for row in hashtag_rows], page_params) + trend_page = paginate_items([HashtagTrend(**row) for row in trend_rows], page_params) + return HashtagStatsResponse( + count=len(hashtag_page.items), start=start, end=end, hashtag=normalized_hashtag, - tags=tags, + interval=interval, + limit=limit, + offset=offset, + pagination=hashtag_page, + hashtags=hashtag_page.items, + trends=trend_page.items, + ) + + +class EditorStatsController(Controller): + path = "/editor-stats" + + @get() + async def get_editor_stats( + self, + start: Annotated[ + datetime | None, Parameter(description="Inclusive UTC lower bound (ISO 8601). If omitted, no lower bound.") + ] = None, + end: Annotated[ + datetime | None, + Parameter(description="Exclusive UTC upper bound (ISO 8601). Defaults to now if start is set."), + ] = None, + include_version: Annotated[ + bool, + Parameter(description="Include editor versions instead of grouping results by editor family."), + ] = False, + limit: Annotated[int, Parameter(ge=1, le=1000, description="Page size (1-1000).")] = 100, + offset: Annotated[int, Parameter(ge=0, description="Page offset.")] = 0, + ) -> EditorStatsResponse: + start, end = resolve_optional_window(start, end) + page_params = PaginationParams(limit=limit, offset=offset) + rows = await fetch_editor_stats( + start=start, + end=end, + include_version=include_version, + limit=page_params.query_limit, + offset=page_params.offset, + ) + page = paginate_items([EditorStat(**row) for row in rows], page_params) + return EditorStatsResponse( + count=len(page.items), + start=start, + end=end, + include_version=include_version, + limit=limit, + offset=offset, + pagination=page, + editors=page.items, + ) + + +class MapController(Controller): + path = "/map" + + @get() + async def get_map( + self, + start: Annotated[ + datetime | None, Parameter(description="Inclusive UTC lower bound (ISO 8601). If omitted, no lower bound.") + ] = None, + end: Annotated[ + datetime | None, + Parameter(description="Exclusive UTC upper bound (ISO 8601). Defaults to now if start is set."), + ] = None, + hashtag: Annotated[ + list[str] | None, Parameter(description="Filter to changesets carrying any of these hashtags. Repeatable.") + ] = None, + limit: Annotated[int, Parameter(ge=1, le=1000, description="Page size (1-1000).")] = 500, + offset: Annotated[int, Parameter(ge=0, description="Page offset.")] = 0, + ) -> MapFeatureCollection: + start, end = resolve_optional_window(start, end) + normalized_hashtag = normalize_hashtags(hashtag) + page_params = PaginationParams(limit=limit, offset=offset) + feature_collection = await fetch_map_changes( + start=start, + end=end, + hashtag=normalized_hashtag, + limit=page_params.query_limit, + offset=page_params.offset, + ) + page = paginate_items(feature_collection["features"], page_params) + return MapFeatureCollection( + count=len(page.items), limit=limit, offset=offset, - users=users, + pagination=page, + features=page.items, ) -v1_router = Router(path="/api/v1", route_handlers=[StatsController]) +v1_router = Router( + path="/api/v1", + route_handlers=[StatsController, HashtagStatsController, EditorStatsController, MapController], +) diff --git a/api/schemas.py b/api/schemas.py index 8ec604f..1066f8d 100644 --- a/api/schemas.py +++ b/api/schemas.py @@ -1,5 +1,7 @@ from datetime import datetime +from typing import Any, Literal +from litestar.pagination import OffsetPagination from pydantic import BaseModel, Field @@ -9,6 +11,12 @@ class TagValueStats(BaseModel): len: float | None = None +class TagKeyStats(BaseModel): + c: int = 0 + m: int = 0 + len: float | None = None + + class UserStat(BaseModel): uid: int name: str @@ -27,7 +35,7 @@ class UserStat(BaseModel): map_changes: int rank: int hashtags: list[str] = Field(default_factory=list) - tag_stats: dict[str, dict[str, TagValueStats]] | None = None + tag_stats: dict[str, TagKeyStats | dict[str, TagValueStats]] | None = None class UserStatsResponse(BaseModel): @@ -36,11 +44,75 @@ class UserStatsResponse(BaseModel): end: datetime | None hashtag: list[str] | None tags: bool + tag_mode: Literal["none", "keys", "all"] limit: int offset: int + pagination: OffsetPagination[UserStat] users: list[UserStat] +class MapFeatureCollection(BaseModel): + type: str = "FeatureCollection" + count: int + limit: int + offset: int + pagination: OffsetPagination[dict[str, Any]] + features: list[dict[str, Any]] + + +class HashtagStat(BaseModel): + hashtag: str + changesets: int + users: int + map_changes: int + rank: int + + +class HashtagTrend(BaseModel): + period_start: datetime + hashtag: str + changesets: int + users: int + map_changes: int + + +class HashtagTrendsResponse(BaseModel): + interval: str + trends: list[HashtagTrend] + + +class HashtagStatsResponse(BaseModel): + count: int + start: datetime + end: datetime + hashtag: list[str] | None + interval: str + limit: int + offset: int + pagination: OffsetPagination[HashtagStat] + hashtags: list[HashtagStat] + trends: list[HashtagTrend] + + +class EditorStat(BaseModel): + editor: str + changesets: int + users: int + map_changes: int + rank: int + + +class EditorStatsResponse(BaseModel): + count: int + start: datetime | None + end: datetime | None + include_version: bool + limit: int + offset: int + pagination: OffsetPagination[EditorStat] + editors: list[EditorStat] + + class HealthResponse(BaseModel): status: str last_seq: int | None diff --git a/docs/infra.md b/docs/infra.md index 3a9ce46..4a9a81a 100644 --- a/docs/infra.md +++ b/docs/infra.md @@ -109,6 +109,33 @@ docker image prune -af && docker builder prune -af ```text GET /health -GET /api/v1/stats?start=&end=[&hashtag=][&tags=true|false][&limit=N][&offset=N] +GET /api/v1/stats?start=&end=[&hashtag=][&tags=true|false][&tag_mode=keys|all][&limit=N][&offset=N] +GET /api/v1/hashtag-stats?start=&end=[&hashtag=][&interval=day|week|month][&limit=N][&offset=N] +GET /api/v1/editor-stats?start=&end=[&include_version=true|false][&limit=N][&offset=N] +GET /api/v1/map?start=&end=[&hashtag=][&limit=N][&offset=N] GET /docs/swagger ``` + +`tags=true&tag_mode=keys` (the default) returns compact per-key totals. Use +`tag_mode=all` to opt in to the full key/value breakdown, or `tags=false` to skip +JSONB expansion for cheaper responses. Editor stats group versions into editor +families by default; use `include_version=true` for full version strings. Collection +responses include a Litestar `OffsetPagination` object with `items`, `limit`, +`offset`, and `total`. Map +features use centroid Point geometry only, ready for clustering and heatmaps. + +## Run the API standalone (without compose) + +```bash +uv run osmsg --last day --format psql --psql-dsn "$DATABASE_URL" --name api_last_day +uv run --group api litestar --app api.app:app run --host 0.0.0.0 --port 8000 +``` + +## Volumes + +| Volume | Contents | +| --- | --- | +| `pgdata` | Postgres data | +| `osmsg-data` | DuckDB state files + parquet output | +| `osmsg-cache` | Downloaded replication diff cache | +| `caddy-data` | TLS certificates | diff --git a/osmsg/export/psql.py b/osmsg/export/psql.py index a98241e..0856c29 100644 --- a/osmsg/export/psql.py +++ b/osmsg/export/psql.py @@ -6,9 +6,13 @@ from ..pg_schema import PG_SCHEMA _BULK_INDEXES = [ - ("idx_changesets_created_at", "CREATE INDEX idx_changesets_created_at ON changesets (created_at)"), - ("idx_changesets_geom", "CREATE INDEX idx_changesets_geom ON changesets USING GIST (geom)"), - ("idx_changeset_stats_uid", "CREATE INDEX idx_changeset_stats_uid ON changeset_stats (uid)"), + ("idx_changesets_created_at", "CREATE INDEX idx_changesets_created_at ON changesets USING BTREE (created_at)"), + ( + "idx_changesets_bbox", + "CREATE INDEX idx_changesets_bbox ON changesets USING GIST " + "(box(point(min_lon, min_lat), point(max_lon, max_lat)))", + ), + ("idx_changeset_stats_uid", "CREATE INDEX idx_changeset_stats_uid ON changeset_stats USING BTREE (uid)"), ] _BULK_FKS = [ ("changesets", "changesets_uid_fkey", "FOREIGN KEY (uid) REFERENCES users (uid)"), @@ -21,6 +25,30 @@ ] +def _changeset_bbox_select(conn: duckdb.DuckDBPyConnection) -> str: + columns = { + row[0] + for row in conn.execute( + "SELECT column_name FROM information_schema.columns WHERE table_name = 'changesets'" + ).fetchall() + } + if {"min_lon", "min_lat", "max_lon", "max_lat"}.issubset(columns): + return "min_lon, min_lat, max_lon, max_lat" + if "geom" in columns: + return """ + CASE WHEN geom IS NOT NULL THEN ST_XMin(geom) END AS min_lon, + CASE WHEN geom IS NOT NULL THEN ST_YMin(geom) END AS min_lat, + CASE WHEN geom IS NOT NULL THEN ST_XMax(geom) END AS max_lon, + CASE WHEN geom IS NOT NULL THEN ST_YMax(geom) END AS max_lat + """ + return """ + NULL::DOUBLE AS min_lon, + NULL::DOUBLE AS min_lat, + NULL::DOUBLE AS max_lon, + NULL::DOUBLE AS max_lat + """ + + _BULK_COMMIT_CHUNKS = 32 @@ -37,15 +65,29 @@ def _pg_has_history(conn: duckdb.DuckDBPyConnection) -> bool: def _push_changesets(conn: duckdb.DuckDBPyConnection, where: str = "") -> None: # Newer non-NULL wins, NULL never downgrades (mirrors the DuckDB-side merge). + bbox_select = _changeset_bbox_select(conn) conn.execute( f""" - INSERT INTO pg_target.changesets AS c (changeset_id, uid, created_at, hashtags, editor, geom) - SELECT changeset_id, uid, created_at, hashtags, editor, geom FROM changesets {where} + INSERT INTO pg_target.changesets AS c ( + changeset_id, uid, created_at, hashtags, editor, + min_lon, min_lat, max_lon, max_lat + ) + SELECT + changeset_id, + uid, + created_at, + hashtags, + editor, + {bbox_select} + FROM changesets {where} ON CONFLICT (changeset_id) DO UPDATE SET created_at = COALESCE(EXCLUDED.created_at, c.created_at), hashtags = COALESCE(EXCLUDED.hashtags, c.hashtags), editor = COALESCE(EXCLUDED.editor, c.editor), - geom = COALESCE(EXCLUDED.geom, c.geom) + min_lon = COALESCE(EXCLUDED.min_lon, c.min_lon), + min_lat = COALESCE(EXCLUDED.min_lat, c.min_lat), + max_lon = COALESCE(EXCLUDED.max_lon, c.max_lon), + max_lat = COALESCE(EXCLUDED.max_lat, c.max_lat) """ ) diff --git a/osmsg/pg_schema.py b/osmsg/pg_schema.py index e22aec4..c1826c9 100644 --- a/osmsg/pg_schema.py +++ b/osmsg/pg_schema.py @@ -1,5 +1,4 @@ PG_SCHEMA = """ -CREATE EXTENSION IF NOT EXISTS postgis; CREATE TABLE IF NOT EXISTS users ( uid BIGINT PRIMARY KEY, username TEXT NOT NULL @@ -10,10 +9,17 @@ created_at TIMESTAMPTZ, hashtags TEXT[], editor TEXT, - geom GEOMETRY(POLYGON) + min_lon DOUBLE PRECISION, + min_lat DOUBLE PRECISION, + max_lon DOUBLE PRECISION, + max_lat DOUBLE PRECISION +); +CREATE INDEX IF NOT EXISTS idx_changesets_created_at ON changesets USING BTREE (created_at); +CREATE INDEX IF NOT EXISTS idx_changesets_hashtags ON changesets USING GIN (hashtags); +CREATE INDEX IF NOT EXISTS idx_changesets_editor ON changesets USING BTREE (editor); +CREATE INDEX IF NOT EXISTS idx_changesets_bbox ON changesets USING GIST ( + box(point(min_lon, min_lat), point(max_lon, max_lat)) ); -CREATE INDEX IF NOT EXISTS idx_changesets_created_at ON changesets(created_at); -CREATE INDEX IF NOT EXISTS idx_changesets_geom ON changesets USING GIST (geom); CREATE TABLE IF NOT EXISTS changeset_stats ( changeset_id BIGINT NOT NULL REFERENCES changesets(changeset_id), seq_id BIGINT NOT NULL, @@ -32,8 +38,8 @@ tag_stats JSONB, PRIMARY KEY (seq_id, changeset_id) ); -CREATE INDEX IF NOT EXISTS idx_changeset_stats_uid ON changeset_stats(uid); -CREATE INDEX IF NOT EXISTS idx_changeset_stats_changeset_id ON changeset_stats(changeset_id); +CREATE INDEX IF NOT EXISTS idx_changeset_stats_uid ON changeset_stats USING BTREE (uid); +CREATE INDEX IF NOT EXISTS idx_changeset_stats_changeset_id ON changeset_stats USING BTREE (changeset_id); CREATE TABLE IF NOT EXISTS state ( source_url TEXT PRIMARY KEY, last_seq BIGINT NOT NULL, diff --git a/scripts/check_api_query_performance.py b/scripts/check_api_query_performance.py new file mode 100644 index 0000000..d316db5 --- /dev/null +++ b/scripts/check_api_query_performance.py @@ -0,0 +1,225 @@ +import argparse +import asyncio +import os +from datetime import UTC, datetime, timedelta +from typing import Any + +import asyncpg + +from api.queries import _editor_stats_sql, _hashtag_stats_sql, _hashtag_trends_sql + + +def parse_dt(value: str) -> datetime: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=UTC) + return parsed + + +def normalize_hashtags(values: list[str]) -> list[str]: + return ["#" + value.strip().lstrip("#") for value in values if value.strip()] + + +async def explain(conn: asyncpg.Connection, name: str, sql: str, params: list[Any]) -> None: + plan_rows = await conn.fetch(f"EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT) {sql}", *params) + print(f"\n=== {name} ===") + for row in plan_rows: + print(row["QUERY PLAN"]) + + +async def table_counts(conn: asyncpg.Connection) -> None: + print("=== table counts ===") + for table in ("users", "changesets", "changeset_stats"): + count = await conn.fetchval(f"SELECT COUNT(*) FROM {table}") + print(f"{table}: {count:,}") + row = await conn.fetchrow( + "SELECT MIN(created_at) AS min_created_at, MAX(created_at) AS max_created_at FROM changesets" + ) + print(f"changesets.created_at: {row['min_created_at']} -> {row['max_created_at']}") + + +async def create_synthetic_tables(conn: asyncpg.Connection, rows: int) -> None: + await conn.execute( + """ + CREATE TEMP TABLE users ( + uid BIGINT PRIMARY KEY, + username TEXT NOT NULL + ) ON COMMIT PRESERVE ROWS + """ + ) + await conn.execute( + """ + CREATE TEMP TABLE changesets ( + changeset_id BIGINT PRIMARY KEY, + uid BIGINT NOT NULL, + created_at TIMESTAMPTZ, + hashtags TEXT[], + editor TEXT + ) ON COMMIT PRESERVE ROWS + """ + ) + await conn.execute( + """ + CREATE TEMP TABLE changeset_stats ( + changeset_id BIGINT NOT NULL, + seq_id BIGINT NOT NULL, + uid BIGINT NOT NULL, + nodes_created INTEGER DEFAULT 0, + nodes_modified INTEGER DEFAULT 0, + nodes_deleted INTEGER DEFAULT 0, + ways_created INTEGER DEFAULT 0, + ways_modified INTEGER DEFAULT 0, + ways_deleted INTEGER DEFAULT 0, + rels_created INTEGER DEFAULT 0, + rels_modified INTEGER DEFAULT 0, + rels_deleted INTEGER DEFAULT 0, + poi_created INTEGER DEFAULT 0, + poi_modified INTEGER DEFAULT 0, + tag_stats JSONB, + PRIMARY KEY (seq_id, changeset_id) + ) ON COMMIT PRESERVE ROWS + """ + ) + await conn.execute( + """ + INSERT INTO users (uid, username) + SELECT uid, 'user_' || uid::text + FROM generate_series(1, LEAST($1::bigint, 100000::bigint)) AS uid + """, + rows, + ) + await conn.execute( + """ + INSERT INTO changesets (changeset_id, uid, created_at, hashtags, editor) + SELECT + id, + (id % LEAST($1::bigint, 100000::bigint)) + 1, + TIMESTAMPTZ '2026-05-01T00:00:00Z' + ((id % 43200) * INTERVAL '1 minute'), + CASE + WHEN id % 10 = 0 THEN ARRAY['#maproulette', '#tomtom'] + WHEN id % 7 = 0 THEN ARRAY['#hotosm'] + WHEN id % 5 = 0 THEN ARRAY['#osmnepal', '#buildings'] + ELSE ARRAY[]::TEXT[] + END, + CASE + WHEN id % 4 = 0 THEN 'iD 2.34.0' + WHEN id % 4 = 1 THEN 'JOSM' + WHEN id % 4 = 2 THEN 'StreetComplete' + ELSE NULL + END + FROM generate_series(1, $1::bigint) AS id + """, + rows, + ) + await conn.execute( + """ + INSERT INTO changeset_stats ( + changeset_id, seq_id, uid, nodes_created, nodes_modified, nodes_deleted, + ways_created, ways_modified, ways_deleted, rels_created, rels_modified, rels_deleted, + poi_created, poi_modified + ) + SELECT + id, + id, + (id % LEAST($1::bigint, 100000::bigint)) + 1, + (id % 97)::integer, + (id % 13)::integer, + (id % 3)::integer, + (id % 11)::integer, + (id % 5)::integer, + (id % 2)::integer, + (id % 7)::integer, + (id % 3)::integer, + (id % 2)::integer, + (id % 17)::integer, + (id % 5)::integer + FROM generate_series(1, $1::bigint) AS id + """, + rows, + ) + await ensure_indexes(conn) + await conn.execute("ANALYZE users") + await conn.execute("ANALYZE changesets") + await conn.execute("ANALYZE changeset_stats") + + +async def ensure_indexes(conn: asyncpg.Connection) -> None: + statements = [ + "CREATE INDEX IF NOT EXISTS idx_changesets_created_at ON changesets USING BTREE (created_at)", + "CREATE INDEX IF NOT EXISTS idx_changesets_hashtags ON changesets USING GIN (hashtags)", + "CREATE INDEX IF NOT EXISTS idx_changesets_editor ON changesets USING BTREE (editor)", + "CREATE INDEX IF NOT EXISTS idx_changesets_bbox ON changesets USING GIST " + "(box(point(min_lon, min_lat), point(max_lon, max_lat)))", + "CREATE INDEX IF NOT EXISTS idx_changeset_stats_changeset_id ON changeset_stats USING BTREE (changeset_id)", + ] + for statement in statements: + await conn.execute(statement) + + +async def main() -> None: + parser = argparse.ArgumentParser(description="Run EXPLAIN ANALYZE for API analytics queries.") + parser.add_argument("--database-url", default=os.environ.get("DATABASE_URL"), help="PostgreSQL connection URL.") + parser.add_argument("--start", type=parse_dt, help="Inclusive UTC lower bound, e.g. 2026-05-01T00:00:00Z.") + parser.add_argument("--end", type=parse_dt, help="Exclusive UTC upper bound, e.g. 2026-05-02T00:00:00Z.") + parser.add_argument("--days", type=int, default=30, help="Window size when start/end are omitted.") + parser.add_argument("--hashtag", action="append", default=[], help="Optional hashtag filter. Repeatable.") + parser.add_argument("--interval", choices=("day", "week", "month"), default="day", help="Trend bucket.") + parser.add_argument("--limit", type=int, default=100, help="Limit used by leaderboard queries.") + parser.add_argument("--offset", type=int, default=0, help="Offset used by leaderboard queries.") + parser.add_argument("--ensure-indexes", action="store_true", help="Create analytics indexes before measuring.") + parser.add_argument("--synthetic-rows", type=int, help="Use temporary generated tables with this many rows.") + args = parser.parse_args() + + if not args.database_url: + raise SystemExit("DATABASE_URL is required. Set it in the environment or pass --database-url.") + + end = args.end or datetime.now(tz=UTC) + start = args.start or (end - timedelta(days=args.days)) + if start >= end: + raise SystemExit("start must be before end") + + hashtags = normalize_hashtags(args.hashtag) + + conn = await asyncpg.connect(args.database_url) + try: + if args.synthetic_rows: + print(f"Creating temporary synthetic dataset with {args.synthetic_rows:,} rows...") + await create_synthetic_tables(conn, args.synthetic_rows) + if args.ensure_indexes: + await ensure_indexes(conn) + await table_counts(conn) + + hashtag_stats_params: list[Any] = [start, end] + if hashtags: + hashtag_stats_params.append(hashtags) + hashtag_stats_params.extend([args.limit, args.offset]) + await explain( + conn, + "hashtag stats", + _hashtag_stats_sql(filter_dates=True, filter_hashtags=bool(hashtags)), + hashtag_stats_params, + ) + + trend_params: list[Any] = [start, end] + if hashtags: + trend_params.append(hashtags) + trend_params.extend([args.interval, args.limit, args.offset]) + await explain( + conn, + "hashtag trends", + _hashtag_trends_sql(filter_hashtags=bool(hashtags)), + trend_params, + ) + + await explain( + conn, + "editor stats", + _editor_stats_sql(filter_dates=True, include_version=False), + [start, end, args.limit, args.offset], + ) + finally: + await conn.close() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/test_api.py b/tests/test_api.py index e4a4106..03dc705 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -6,11 +6,13 @@ import pytest from litestar import Litestar +from litestar.config.cors import CORSConfig from litestar.testing import TestClient from api import app as api_app from api.app import health from api.db import close_pool, ensure_schema, open_pool +from api.pagination import PaginationParams, paginate_items from api.pg_schema import PG_SCHEMA as API_PG_SCHEMA from api.routers.v1 import normalize_hashtags, v1_router from osmsg.export.psql import to_psql @@ -28,6 +30,10 @@ def test_api_exposes_only_active_public_routes(): assert "/health" in paths assert "/api/v1/stats" in paths + assert "/api/v1/hashtag-stats" in paths + assert "/api/v1/editor-stats" in paths + assert "/api/v1/map" in paths + assert "/api/v1/hashtag-trends" not in paths assert "/api/v1/stats/summary" not in paths assert "/api/v1/stats/timeseries" not in paths @@ -56,18 +62,54 @@ def test_normalize_hashtags_dedupes_case_insensitively(): assert normalize_hashtags(["maproulette", "#MapRoulette", "#roads"]) == ["#maproulette", "#roads"] -def _stats_app(monkeypatch, fake_fetch): - monkeypatch.setattr(v1_module, "fetch_user_stats", fake_fetch) +def test_pagination_model_fetches_one_extra_row_for_next_page(): + params = PaginationParams(limit=2, offset=4) + assert params.query_limit == 3 + + page = paginate_items(["a", "b", "c"], params) + + assert page.items == ["a", "b"] + assert page.limit == 2 + assert page.offset == 4 + assert page.total == 7 + + +def test_app_cors_allows_browser_preflight(): + app = Litestar( + route_handlers=[v1_router], + cors_config=CORSConfig(allow_origins=["*"], allow_methods=["GET", "OPTIONS"], allow_headers=["*"]), + ) + with TestClient(app) as client: + response = client.options( + "/api/v1/stats", + headers={ + "Origin": "http://localhost:5173", + "Access-Control-Request-Method": "GET", + "Access-Control-Request-Headers": "content-type", + }, + ) + + assert response.status_code == 204 + assert response.headers["access-control-allow-origin"] == "*" + + +def _v1_app(monkeypatch, **fakes): + for name, fake in fakes.items(): + monkeypatch.setattr(v1_module, name, fake) return Litestar(route_handlers=[v1_router]) +def _stats_app(monkeypatch, fake_fetch): + return _v1_app(monkeypatch, fetch_user_stats=fake_fetch) + + def test_user_stats_endpoint_returns_expected_response(monkeypatch): - async def fake_fetch_user_stats(*, start, end, hashtag, tags, limit, offset): + async def fake_fetch_user_stats(*, start, end, hashtag, tag_mode, limit, offset): assert start.isoformat() == "2026-05-01T00:00:00+00:00" assert end.isoformat() == "2026-05-02T00:00:00+00:00" assert hashtag == ["#mapathon", "#roads"] - assert tags is True - assert limit == 1 + assert tag_mode == "keys" + assert limit == 2 assert offset == 0 return [ { @@ -88,7 +130,7 @@ async def fake_fetch_user_stats(*, start, end, hashtag, tags, limit, offset): "map_changes": 58, "rank": 1, "hashtags": ["#mapathon", "#roads"], - "tag_stats": {"building": {"yes": {"c": 3, "m": 0}}}, + "tag_stats": {"building": {"c": 3, "m": 0}}, } ] @@ -111,8 +153,36 @@ async def fake_fetch_user_stats(*, start, end, hashtag, tags, limit, offset): "end": "2026-05-02T00:00:00Z", "hashtag": ["#mapathon", "#roads"], "tags": True, + "tag_mode": "keys", "limit": 1, "offset": 0, + "pagination": { + "items": [ + { + "uid": 10, + "name": "alice", + "changesets": 2, + "nodes_create": 40, + "nodes_modify": 5, + "nodes_delete": 0, + "ways_create": 12, + "ways_modify": 1, + "ways_delete": 0, + "rels_create": 0, + "rels_modify": 0, + "rels_delete": 0, + "poi_create": 5, + "poi_modify": 1, + "map_changes": 58, + "rank": 1, + "hashtags": ["#mapathon", "#roads"], + "tag_stats": {"building": {"c": 3, "m": 0, "len": None}}, + } + ], + "limit": 1, + "offset": 0, + "total": 1, + }, "users": [ { "uid": 10, @@ -132,7 +202,7 @@ async def fake_fetch_user_stats(*, start, end, hashtag, tags, limit, offset): "map_changes": 58, "rank": 1, "hashtags": ["#mapathon", "#roads"], - "tag_stats": {"building": {"yes": {"c": 3, "m": 0, "len": None}}}, + "tag_stats": {"building": {"c": 3, "m": 0, "len": None}}, } ], } @@ -153,8 +223,8 @@ async def fake_fetch_user_stats(**kwargs): def test_user_stats_endpoint_tags_false_drops_tag_stats(monkeypatch): - async def fake_fetch_user_stats(*, tags, **_kwargs): - assert tags is False + async def fake_fetch_user_stats(*, tag_mode, **_kwargs): + assert tag_mode == "none" return [ { "uid": 10, @@ -184,20 +254,293 @@ async def fake_fetch_user_stats(*, tags, **_kwargs): assert response.status_code == 200 body = response.json() assert body["tags"] is False + assert body["tag_mode"] == "none" assert body["users"][0]["tag_stats"] is None +def test_user_stats_endpoint_all_tag_values_are_opt_in(monkeypatch): + async def fake_fetch_user_stats(*, tag_mode, **_kwargs): + assert tag_mode == "all" + return [] + + with TestClient(_stats_app(monkeypatch, fake_fetch_user_stats)) as client: + response = client.get("/api/v1/stats", params={"tag_mode": "all"}) + + assert response.status_code == 200 + assert response.json()["tag_mode"] == "all" + + +def test_hashtag_stats_endpoint_returns_expected_response(monkeypatch): + async def fake_fetch_hashtag_stats(*, start, end, hashtag, limit, offset): + assert start.isoformat() == "2026-05-01T00:00:00+00:00" + assert end.isoformat() == "2026-05-02T00:00:00+00:00" + assert hashtag == ["#mapathon"] + assert limit == 3 + assert offset == 0 + return [ + {"hashtag": "#mapathon", "changesets": 4, "users": 3, "map_changes": 100, "rank": 1}, + {"hashtag": "#roads", "changesets": 2, "users": 2, "map_changes": 25, "rank": 2}, + ] + + async def fake_fetch_hashtag_trends(*, start, end, interval, hashtag, limit, offset): + assert start.isoformat() == "2026-05-01T00:00:00+00:00" + assert end.isoformat() == "2026-05-02T00:00:00+00:00" + assert interval == "day" + assert hashtag == ["#mapathon"] + assert limit == 3 + assert offset == 0 + return [ + { + "period_start": datetime(2026, 5, 1, tzinfo=UTC), + "hashtag": "#mapathon", + "changesets": 4, + "users": 3, + "map_changes": 100, + } + ] + + with TestClient( + _v1_app( + monkeypatch, + fetch_hashtag_stats=fake_fetch_hashtag_stats, + fetch_hashtag_trends=fake_fetch_hashtag_trends, + ) + ) as client: + response = client.get( + "/api/v1/hashtag-stats", + params={ + "start": "2026-05-01T00:00:00Z", + "end": "2026-05-02T00:00:00Z", + "hashtag": "mapathon", + "limit": "2", + }, + ) + + assert response.status_code == 200 + assert response.json() == { + "count": 2, + "start": "2026-05-01T00:00:00Z", + "end": "2026-05-02T00:00:00Z", + "hashtag": ["#mapathon"], + "interval": "day", + "limit": 2, + "offset": 0, + "pagination": { + "items": [ + {"hashtag": "#mapathon", "changesets": 4, "users": 3, "map_changes": 100, "rank": 1}, + {"hashtag": "#roads", "changesets": 2, "users": 2, "map_changes": 25, "rank": 2}, + ], + "limit": 2, + "offset": 0, + "total": 2, + }, + "hashtags": [ + {"hashtag": "#mapathon", "changesets": 4, "users": 3, "map_changes": 100, "rank": 1}, + {"hashtag": "#roads", "changesets": 2, "users": 2, "map_changes": 25, "rank": 2}, + ], + "trends": [ + { + "period_start": "2026-05-01T00:00:00Z", + "hashtag": "#mapathon", + "changesets": 4, + "users": 3, + "map_changes": 100, + }, + ], + } + + +def test_hashtag_stats_endpoint_rejects_invalid_interval(monkeypatch): + async def fake_fetch_hashtag_trends(**kwargs): + raise AssertionError("fetch_hashtag_trends should not be called") + + with TestClient(_v1_app(monkeypatch, fetch_hashtag_trends=fake_fetch_hashtag_trends)) as client: + response = client.get("/api/v1/hashtag-stats", params={"interval": "hour"}) + + assert response.status_code == 400 + assert response.json()["detail"] == "interval must be one of: day, week, month" + + +def test_editor_stats_endpoint_returns_expected_response(monkeypatch): + async def fake_fetch_editor_stats(*, start, end, include_version, limit, offset): + assert start is None + assert end is None + assert include_version is False + assert limit == 3 + assert offset == 0 + return [ + {"editor": "iD 2.34.0", "changesets": 10, "users": 5, "map_changes": 500, "rank": 1}, + {"editor": "JOSM", "changesets": 4, "users": 2, "map_changes": 120, "rank": 2}, + ] + + with TestClient(_v1_app(monkeypatch, fetch_editor_stats=fake_fetch_editor_stats)) as client: + response = client.get("/api/v1/editor-stats", params={"limit": "2"}) + + assert response.status_code == 200 + assert response.json() == { + "count": 2, + "start": None, + "end": None, + "include_version": False, + "limit": 2, + "offset": 0, + "pagination": { + "items": [ + {"editor": "iD 2.34.0", "changesets": 10, "users": 5, "map_changes": 500, "rank": 1}, + {"editor": "JOSM", "changesets": 4, "users": 2, "map_changes": 120, "rank": 2}, + ], + "limit": 2, + "offset": 0, + "total": 2, + }, + "editors": [ + {"editor": "iD 2.34.0", "changesets": 10, "users": 5, "map_changes": 500, "rank": 1}, + {"editor": "JOSM", "changesets": 4, "users": 2, "map_changes": 120, "rank": 2}, + ], + } + + +def test_editor_stats_endpoint_versions_are_opt_in(monkeypatch): + async def fake_fetch_editor_stats(*, include_version, **_kwargs): + assert include_version is True + return [] + + with TestClient(_v1_app(monkeypatch, fetch_editor_stats=fake_fetch_editor_stats)) as client: + response = client.get("/api/v1/editor-stats", params={"include_version": "true"}) + + assert response.status_code == 200 + assert response.json()["include_version"] is True + + +def test_map_endpoint_returns_geojson_feature_collection(monkeypatch): + async def fake_fetch_map_changes(*, start, end, hashtag, limit, offset): + assert start.isoformat() == "2026-05-01T00:00:00+00:00" + assert end.isoformat() == "2026-05-02T00:00:00+00:00" + assert hashtag == ["#mapathon"] + assert limit == 2 + assert offset == 0 + return { + "type": "FeatureCollection", + "features": [ + { + "type": "Feature", + "geometry": {"type": "Point", "coordinates": [85.5, 27.5]}, + "properties": {"changeset_id": 1, "name": "alice", "map_changes": 10}, + } + ], + } + + with TestClient(_v1_app(monkeypatch, fetch_map_changes=fake_fetch_map_changes)) as client: + response = client.get( + "/api/v1/map", + params={ + "start": "2026-05-01T00:00:00Z", + "end": "2026-05-02T00:00:00Z", + "hashtag": "mapathon", + "limit": "1", + }, + ) + + assert response.status_code == 200 + assert response.json()["count"] == 1 + assert response.json()["features"][0]["geometry"] == {"type": "Point", "coordinates": [85.5, 27.5]} + assert response.json()["features"][0]["properties"] == {"changeset_id": 1, "name": "alice", "map_changes": 10} + + def test_user_stats_sql_omits_tag_ctes_when_tags_false(): from api.queries import _user_stats_sql - sql_with = _user_stats_sql(filter_dates=False, filter_hashtags=False, include_tags=True) - sql_without = _user_stats_sql(filter_dates=False, filter_hashtags=False, include_tags=False) + sql_with = _user_stats_sql(filter_dates=False, filter_hashtags=False, tag_mode="all") + sql_without = _user_stats_sql(filter_dates=False, filter_hashtags=False, tag_mode="none") assert "tag_per_user" in sql_with + assert "COALESCE(tpu.tag_stats, '{}'::jsonb) AS tag_stats" in sql_with assert "tag_per_user" not in sql_without assert "NULL::jsonb AS tag_stats" in sql_without assert "user_hashtags" in sql_without +def test_user_stats_sql_aggregates_tag_keys_by_default(): + from api.queries import _user_stats_sql + + sql = _user_stats_sql(filter_dates=False, filter_hashtags=False, tag_mode="keys") + assert "jsonb_object_agg" in sql + assert "SUM(COALESCE((tv.value->>'c')::bigint, 0))" in sql + assert "tag_val" not in sql + + +def test_hashtag_stats_sql_uses_array_overlap_and_lateral_unnest(): + from api.queries import _hashtag_stats_sql + + sql = _hashtag_stats_sql(filter_dates=True, filter_hashtags=True) + assert "CROSS JOIN LATERAL UNNEST(cs.hashtags)" in sql + assert "cs.hashtags && $3::TEXT[]" in sql + assert "ht.hashtag = ANY($3::TEXT[])" in sql + assert "LIMIT $4 OFFSET $5" in sql + + +def test_hashtag_trends_sql_uses_bounded_date_range(): + from api.queries import _hashtag_trends_sql + + sql = _hashtag_trends_sql(filter_hashtags=False) + assert "DATE_TRUNC($3, cs.created_at)" in sql + assert "cs.created_at >= $1" in sql + assert "cs.created_at < $2" in sql + assert "LIMIT $4 OFFSET $5" in sql + + +def test_hashtag_trends_sql_filters_unnested_hashtags_when_requested(): + from api.queries import _hashtag_trends_sql + + sql = _hashtag_trends_sql(filter_hashtags=True) + assert "cs.hashtags && $3::TEXT[]" in sql + assert "ht.hashtag = ANY($3::TEXT[])" in sql + assert "DATE_TRUNC($4, cs.created_at)" in sql + + +def test_map_changes_sql_returns_geojson_geometry(): + from api.queries import _map_changes_sql + + sql = _map_changes_sql( + filter_dates=True, + filter_hashtags=True, + geometry_mode="geom", + ) + assert "ST_AsGeoJSON(ST_Centroid(cs.geom))::TEXT AS geometry" in sql + assert "cs.hashtags && $3::TEXT[]" in sql + assert " AS centroid" not in sql + assert "LIMIT $4 OFFSET $5" in sql + + +def test_map_changes_sql_can_return_bbox_geojson_geometry(): + from api.queries import _map_changes_sql + + sql = _map_changes_sql( + filter_dates=True, + filter_hashtags=False, + geometry_mode="bbox", + ) + assert "'type', 'Point'" in sql + assert "'type', 'Polygon'" not in sql + assert "cs.min_lon, cs.min_lat, cs.max_lon, cs.max_lat" in sql + assert "LIMIT $3 OFFSET $4" in sql + + +def test_editor_stats_sql_groups_blank_editors_as_unknown(): + from api.queries import _editor_stats_sql + + sql = _editor_stats_sql(filter_dates=False, include_version=False) + assert "REGEXP_REPLACE" in sql + assert "GROUP BY editor" in sql + + +def test_editor_stats_sql_can_include_versions(): + from api.queries import _editor_stats_sql + + sql = _editor_stats_sql(filter_dates=False, include_version=True) + assert "COALESCE(NULLIF(cs.editor, ''), 'unknown') AS editor" in sql + assert "REGEXP_REPLACE" not in sql + + def _seed_pg_via_to_psql(fresh_db, populated_db_factory, dsn): populated = populated_db_factory(fresh_db) populated.execute( @@ -269,15 +612,24 @@ def test_live_api_stats_default_returns_dicts_not_strings(live_api_client): assert r.status_code == 200, r.text body = r.json() assert body["tags"] is True + assert body["tag_mode"] == "keys" assert body["count"] == 2 by_name = {u["name"]: u for u in body["users"]} assert isinstance(by_name["alice"]["tag_stats"], dict) assert by_name["alice"]["hashtags"] == ["#mapathon"] assert by_name["bob"]["hashtags"] == [] + assert by_name["alice"]["tag_stats"]["building"]["c"] == 7 + assert by_name["alice"]["tag_stats"]["building"]["m"] == 1 + assert by_name["alice"]["tag_stats"]["highway"]["len"] == 245.7 + assert by_name["bob"]["tag_stats"]["natural"]["c"] == 50 + + +@pytest.mark.network +def test_live_api_stats_all_mode_returns_value_breakdown(live_api_client): + body = live_api_client.get("/api/v1/stats", params={"tag_mode": "all"}).json() + by_name = {u["name"]: u for u in body["users"]} assert by_name["alice"]["tag_stats"]["building"]["yes"]["c"] == 5 - assert by_name["alice"]["tag_stats"]["building"]["yes"]["m"] == 1 assert by_name["alice"]["tag_stats"]["highway"]["residential"]["len"] == 245.7 - assert by_name["bob"]["tag_stats"]["natural"]["tree"]["c"] == 50 @pytest.mark.network @@ -388,6 +740,7 @@ def test_live_api_stats_response_echoes_query(live_api_client): assert body["end"] == "2026-04-03T00:00:00Z" assert body["hashtag"] == ["#mapathon"] assert body["tags"] is True + assert body["tag_mode"] == "keys" assert body["limit"] == 10 assert body["offset"] == 0 @@ -451,7 +804,7 @@ def test_user_stats_sql_no_filter_skips_changesets_join(): """No-filter path must not JOIN the changesets table, orphan stats would be dropped.""" from api.queries import _user_stats_sql - sql = _user_stats_sql(filter_dates=False, filter_hashtags=False, include_tags=False) + sql = _user_stats_sql(filter_dates=False, filter_hashtags=False, tag_mode="none") assert "filtered_changesets" not in sql assert "JOIN filtered_changesets" not in sql assert "stats_scope AS (SELECT * FROM changeset_stats)" in sql @@ -462,8 +815,8 @@ def test_user_stats_sql_filtered_uses_changesets_join(): """Filtered path must scope through changesets so date/hashtag predicates apply.""" from api.queries import _user_stats_sql - sql_dates = _user_stats_sql(filter_dates=True, filter_hashtags=False, include_tags=False) - sql_tags = _user_stats_sql(filter_dates=False, filter_hashtags=True, include_tags=False) + sql_dates = _user_stats_sql(filter_dates=True, filter_hashtags=False, tag_mode="none") + sql_tags = _user_stats_sql(filter_dates=False, filter_hashtags=True, tag_mode="none") for sql in (sql_dates, sql_tags): assert "filtered_changesets" in sql assert "JOIN filtered_changesets" in sql @@ -474,6 +827,10 @@ def test_user_stats_sql_no_unfiltered_fallback_remains(): from api.queries import _user_stats_sql for combo in [(True, True), (True, False), (False, True), (False, False)]: - sql = _user_stats_sql(filter_dates=combo[0], filter_hashtags=combo[1], include_tags=False) + sql = _user_stats_sql( + filter_dates=combo[0], + filter_hashtags=combo[1], + tag_mode="none", + ) assert "NOT EXISTS (SELECT 1 FROM matching_stats)" not in sql assert "enable_unfiltered_fallback" not in sql diff --git a/tests/test_psql_export.py b/tests/test_psql_export.py index 3c849e2..d614529 100644 --- a/tests/test_psql_export.py +++ b/tests/test_psql_export.py @@ -46,24 +46,27 @@ def test_pg_schema_statements_each_parse_with_postgres_extension(): """Each individual CREATE statement is well-formed enough that the postgres extension's parser would accept it, we use DuckDB's own parser as an approximation (DuckDB's CREATE TABLE syntax is compatible).""" - duckdb_clone = ( - PG_SCHEMA.replace("DOUBLE PRECISION", "DOUBLE") - .replace("JSONB", "JSON") - .replace("TEXT", "VARCHAR") - .replace("GEOMETRY(POLYGON)", "GEOMETRY") - ) + duckdb_clone = PG_SCHEMA.replace("DOUBLE PRECISION", "DOUBLE").replace("JSONB", "JSON").replace("TEXT", "VARCHAR") conn = duckdb.connect(":memory:") conn.execute("INSTALL spatial") conn.execute("LOAD spatial") for stmt in [s.strip() for s in duckdb_clone.split(";") if s.strip()]: upper = stmt.upper() - if upper.startswith("CREATE EXTENSION") or "USING GIST" in upper: + if "CREATE INDEX" in upper and "USING" in upper: continue conn.execute(stmt) tables = {r[0] for r in conn.execute("SELECT table_name FROM information_schema.tables").fetchall()} assert {"users", "changesets", "changeset_stats", "state"} <= tables +def test_pg_schema_uses_explicit_postgres_index_types(): + assert "idx_changesets_created_at ON changesets USING BTREE (created_at)" in PG_SCHEMA + assert "idx_changesets_hashtags ON changesets USING GIN (hashtags)" in PG_SCHEMA + assert "idx_changesets_editor ON changesets USING BTREE (editor)" in PG_SCHEMA + assert "idx_changesets_bbox ON changesets USING GIST" in PG_SCHEMA + assert "box(point(min_lon, min_lat), point(max_lon, max_lat))" in PG_SCHEMA + + EXPECTED_USER_STATS = { "alice": {"changesets": 1, "nodes_create": 30, "ways_create": 8, "poi_create": 5, "map_changes": 44}, "bob": {"changesets": 1, "nodes_create": 50, "ways_create": 0, "poi_create": 50, "map_changes": 50}, @@ -323,8 +326,12 @@ def test_to_psql_upgrades_empty_changeset_when_pushed_again(fresh_db, tmp_path): verifier.execute("LOAD postgres") verifier.execute(f"ATTACH '{safe_dsn}' AS pg_src (TYPE postgres, READ_ONLY)") try: - editor, hashtags, has_geom = verifier.execute( - "SELECT editor, hashtags, geom IS NOT NULL FROM pg_src.changesets WHERE changeset_id = 900900" + editor, hashtags, min_lon, min_lat, max_lon, max_lat = verifier.execute( + """ + SELECT editor, hashtags, min_lon, min_lat, max_lon, max_lat + FROM pg_src.changesets + WHERE changeset_id = 900900 + """ ).fetchone() n_stats = verifier.execute( "SELECT COUNT(*) FROM pg_src.changeset_stats WHERE changeset_id = 900900" @@ -335,7 +342,7 @@ def test_to_psql_upgrades_empty_changeset_when_pushed_again(fresh_db, tmp_path): assert editor == "JOSM" assert hashtags == ["#x"] - assert has_geom is True + assert (min_lon, min_lat, max_lon, max_lat) == (10.0, 20.0, 11.0, 21.0) assert n_stats == 1