diff --git a/geocode/model.py b/geocode/model.py index 343dfdc..2de8e0a 100644 --- a/geocode/model.py +++ b/geocode/model.py @@ -1,8 +1,9 @@ """Database model.""" import sqlalchemy +import sqlalchemy.orm.query from geoalchemy2 import Geometry -from sqlalchemy import cast, func +from sqlalchemy import and_, cast, func, or_ from sqlalchemy.dialects import postgresql from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.ext.hybrid import hybrid_property @@ -13,7 +14,7 @@ from sqlalchemy.types import Float, Integer, Numeric, String from .database import session Base = declarative_base() -Base.query = session.query_property() +Base.query = session.query_property() # type:ignore class Polygon(Base): @@ -23,6 +24,7 @@ class Polygon(Base): osm_id = Column(Integer, primary_key=True, autoincrement=False) admin_level = Column(String) + boundary = Column(String) way_area = Column(Float) tags = Column(postgresql.HSTORE) @@ -43,15 +45,20 @@ class Polygon(Base): @classmethod def coords_within( cls, lat: str | float, lon: str | float - ) -> sqlalchemy.orm.query.Query["Polygon"]: + ) -> sqlalchemy.orm.query.Query: # type: ignore """Polygons that contain given coordinates.""" point = func.ST_SetSRID(func.ST_MakePoint(lon, lat), 4326) - q: sqlalchemy.orm.query.Query["Polygon"] = cls.query.filter( - cls.admin_level.isnot(None), - cls.admin_level.regexp_match("^\d+$"), + q = cls.query.filter( # type: ignore + or_( + cls.boundary == "political", + and_( + cls.admin_level.isnot(None), # type: ignore + cls.admin_level.regexp_match(r"^\d+$"), # type: ignore + ), + ), func.ST_Within(point, cls.way), ).order_by(cls.area, cast(cls.admin_level, Integer).desc()) - return q + return q # type: ignore class Scotland(Base): diff --git a/geocode/wikidata.py b/geocode/wikidata.py index 4b4f920..08f4c91 100644 --- a/geocode/wikidata.py +++ b/geocode/wikidata.py @@ -107,6 +107,7 @@ def geosearch(lat: str | float, lon: str | float) -> Row | None: "Q5084": 1, # hamlet "Q515": 2, # city "Q1549591": 3, # big city + "Q589282": 2, # ward or electoral division of the United Kingdom } for row in rows: isa = wd_uri_to_qid(row["isa"]["value"]) diff --git a/lookup.py b/lookup.py index e814377..ece7b2c 100755 --- a/lookup.py +++ b/lookup.py @@ -66,7 +66,6 @@ def lat_lon_to_wikidata(lat: str | float, lon: str | float) -> dict[str, typing. return {"elements": elements, "result": result} admin_level = result.get("admin_level") - assert isinstance(admin_level, int) if not admin_level or admin_level >= 7: return {"elements": elements, "result": result}