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

1import json 

2import logging 

3 

4from sqlalchemy.dialects.postgresql import JSON 

5from sqlalchemy.sql import func 

6 

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 

11 

12logger = logging.getLogger(__name__) 

13 

14 

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 ) 

26 

27 

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 ) 

35 

36 

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) 

41 

42 def GetCommunities(self, request, context, session): 

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

44 

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 ) 

54 

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 ) 

60 

61 return _statement_to_geojson_response(session, statement) 

62 

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 ) 

71 

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 ) 

77 

78 return _statement_to_geojson_response(session, statement)