Coverage for src/couchers/servicers/gis.py: 50%

1import json 

2import logging 

3 

4from sqlalchemy.dialects.postgresql import JSON 

5from sqlalchemy.sql import func 

6 

7from couchers.db import session_scope 

8from couchers.models import Node, Page, PageType, PageVersion, User 

9from couchers.sql import couchers_select as select 

10from proto import gis_pb2_grpc 

11from proto.google.api import httpbody_pb2 

12 

13logger = logging.getLogger(__name__) 

14 

15 

16def _build_geojson_select(statement): 

17 """ 

18 See usages below. 

19 """ 

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

21 return func.json_build_object( 

22 "type", 

23 "FeatureCollection", 

24 "features", 

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

26 ) 

27 

28 

29def _statement_to_geojson_response(session, statement): 

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

31 return httpbody_pb2.HttpBody( 

32 content_type="application/json", 

33 # json.dumps escapes non-ascii characters 

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

35 ) 

36 

37 

38class GIS(gis_pb2_grpc.GISServicer): 

39 def GetUsers(self, request, context): 

40 with session_scope() as session: 

41 statement = select(User.username, User.id, User.geom).where_users_visible(context).where(User.geom != None) 

42 

43 return _statement_to_geojson_response(session, statement) 

44 

45 def GetCommunities(self, request, context): 

46 with session_scope() as session: 

47 statement = select(Node).where(Node.geom != None) 

48 

49 return _statement_to_geojson_response(session, statement) 

50 

51 def GetPlaces(self, request, context): 

52 with session_scope() as session: 

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

54 latest_pages = ( 

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

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

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

58 .group_by(PageVersion.page_id) 

59 .subquery() 

60 ) 

61 

62 statement = ( 

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

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

65 .where(PageVersion.geom != None) 

66 ) 

67 

68 return _statement_to_geojson_response(session, statement) 

69 

70 def GetGuides(self, request, context): 

71 with session_scope() as session: 

72 latest_pages = ( 

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

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

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

76 .group_by(PageVersion.page_id) 

77 .subquery() 

78 ) 

79 

80 statement = ( 

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

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

83 .where(PageVersion.geom != None) 

84 ) 

85 

86 return _statement_to_geojson_response(session, statement)