Coverage for src/couchers/servicers/gis.py: 57%
28 statements
« prev ^ index » next coverage.py v7.5.0, created at 2024-10-04 23:02 +0000
« prev ^ index » next coverage.py v7.5.0, created at 2024-10-04 23:02 +0000
1import json
2import logging
4from sqlalchemy.dialects.postgresql import JSON
5from sqlalchemy.sql import func
7from couchers.models import Node, Page, PageType, PageVersion, User
8from couchers.sql import couchers_select as select
9from proto import gis_pb2_grpc
10from proto.google.api import httpbody_pb2
12logger = logging.getLogger(__name__)
15def _build_geojson_select(statement):
16 """
17 See usages below.
18 """
19 # this is basically a translation of the postgis ST_AsGeoJSON example into sqlalchemy/geoalchemy2
20 return func.json_build_object(
21 "type",
22 "FeatureCollection",
23 "features",
24 func.json_agg(func.ST_AsGeoJSON(statement.subquery(), maxdecimaldigits=5).cast(JSON)),
25 )
28def _statement_to_geojson_response(session, statement):
29 json_dict = session.execute(select(_build_geojson_select(statement))).scalar_one_or_none()
30 return httpbody_pb2.HttpBody(
31 content_type="application/json",
32 # json.dumps escapes non-ascii characters
33 data=json.dumps(json_dict).encode("ascii"),
34 )
37class GIS(gis_pb2_grpc.GISServicer):
38 def GetUsers(self, request, context, session):
39 statement = select(User.username, User.id, User.geom).where_users_visible(context).where(User.geom != None)
40 return _statement_to_geojson_response(session, statement)
42 def GetCommunities(self, request, context, session):
43 return _statement_to_geojson_response(session, select(Node).where(Node.geom != None))
45 def GetPlaces(self, request, context, session):
46 # need to do a subquery here so we get pages without a geom, not just versions without geom
47 latest_pages = (
48 select(func.max(PageVersion.id).label("id"))
49 .join(Page, Page.id == PageVersion.page_id)
50 .where(Page.type == PageType.place)
51 .group_by(PageVersion.page_id)
52 .subquery()
53 )
55 statement = (
56 select(PageVersion.page_id.label("id"), PageVersion.slug.label("slug"), PageVersion.geom)
57 .join(latest_pages, latest_pages.c.id == PageVersion.id)
58 .where(PageVersion.geom != None)
59 )
61 return _statement_to_geojson_response(session, statement)
63 def GetGuides(self, request, context, session):
64 latest_pages = (
65 select(func.max(PageVersion.id).label("id"))
66 .join(Page, Page.id == PageVersion.page_id)
67 .where(Page.type == PageType.guide)
68 .group_by(PageVersion.page_id)
69 .subquery()
70 )
72 statement = (
73 select(PageVersion.page_id.label("id"), PageVersion.slug.label("slug"), PageVersion.geom)
74 .join(latest_pages, latest_pages.c.id == PageVersion.id)
75 .where(PageVersion.geom != None)
76 )
78 return _statement_to_geojson_response(session, statement)