Coverage for app / backend / src / couchers / servicers / gis.py: 81%

1import json 

2import logging 

3from typing import Any 

4 

5from google.protobuf import empty_pb2 

6from sqlalchemy import Function, select 

7from sqlalchemy.dialects.postgresql import JSON 

8from sqlalchemy.orm import Session 

9from sqlalchemy.sql import func 

10from sqlalchemy.sql.selectable import GenerativeSelect 

11 

12from couchers.context import CouchersContext 

13from couchers.materialized_views import ClusteredUser, LiteUser 

14from couchers.models import Node, Page, PageType, PageVersion 

15from couchers.proto import gis_pb2_grpc 

16from couchers.proto.google.api import httpbody_pb2 

17from couchers.sql import users_visible 

18 

19logger = logging.getLogger(__name__) 

20 

21 

22def _build_geojson_select(statement: GenerativeSelect) -> Function[Any]: 

23 """ 

24 See usages below. 

25 """ 

26 # this is basically a translation of the postgis ST_AsGeoJSON example into sqlalchemy/geoalchemy2 

27 return func.json_build_object( 

28 "type", 

29 "FeatureCollection", 

30 "features", 

31 func.json_agg(func.ST_AsGeoJSON(statement.subquery(), maxdecimaldigits=5).cast(JSON)), 

32 ) 

33 

34 

35def _statement_to_geojson_response(session: Session, statement: GenerativeSelect) -> httpbody_pb2.HttpBody: 

36 json_dict = session.execute(select(_build_geojson_select(statement))).scalar_one_or_none() 

37 return httpbody_pb2.HttpBody( 

38 content_type="application/json", 

39 # json.dumps escapes non-ascii characters 

40 data=json.dumps(json_dict).encode("ascii"), 

41 ) 

42 

43 

44class GIS(gis_pb2_grpc.GISServicer): 

45 def GetUsers(self, request: empty_pb2.Empty, context: CouchersContext, session: Session) -> httpbody_pb2.HttpBody: 

46 statement = select(LiteUser.id, LiteUser.geom, LiteUser.has_completed_profile).where( 

47 users_visible(context, table=LiteUser) 

48 ) 

49 return _statement_to_geojson_response(session, statement) 

50 

51 def GetClusteredUsers( 

52 self, request: empty_pb2.Empty, context: CouchersContext, session: Session 

53 ) -> httpbody_pb2.HttpBody: 

54 return _statement_to_geojson_response(session, select(ClusteredUser.geom, ClusteredUser.count)) 

55 

56 def GetCommunities( 

57 self, request: empty_pb2.Empty, context: CouchersContext, session: Session 

58 ) -> httpbody_pb2.HttpBody: 

59 return _statement_to_geojson_response(session, select(Node).where(Node.geom != None)) 

60 

61 def GetPlaces(self, request: empty_pb2.Empty, context: CouchersContext, session: Session) -> httpbody_pb2.HttpBody: 

62 # need to do a subquery here so we get pages without a geom, not just versions without geom 

63 latest_pages = ( 

64 select(func.max(PageVersion.id).label("id")) 

65 .join(Page, Page.id == PageVersion.page_id) 

66 .where(Page.type == PageType.place) 

67 .group_by(PageVersion.page_id) 

68 .subquery() 

69 ) 

70 

71 statement = ( 

72 select(PageVersion.page_id.label("id"), PageVersion.slug.label("slug"), PageVersion.geom) 

73 .join(latest_pages, latest_pages.c.id == PageVersion.id) 

74 .where(PageVersion.geom != None) 

75 ) 

76 

77 return _statement_to_geojson_response(session, statement) 

78 

79 def GetGuides(self, request: empty_pb2.Empty, context: CouchersContext, session: Session) -> httpbody_pb2.HttpBody: 

80 latest_pages = ( 

81 select(func.max(PageVersion.id).label("id")) 

82 .join(Page, Page.id == PageVersion.page_id) 

83 .where(Page.type == PageType.guide) 

84 .group_by(PageVersion.page_id) 

85 .subquery() 

86 ) 

87 

88 statement = ( 

89 select(PageVersion.page_id.label("id"), PageVersion.slug.label("slug"), PageVersion.geom) 

90 .join(latest_pages, latest_pages.c.id == PageVersion.id) 

91 .where(PageVersion.geom != None) 

92 ) 

93 

94 return _statement_to_geojson_response(session, statement)