matcher/api.py

@@ -6,9 +6,12 @@ import typing
 
 import flask
 import geoalchemy2
-import sqlalchemy
+from sqlalchemy import and_, func, or_, text
-from sqlalchemy import and_, or_
+from sqlalchemy.dialects import postgresql
+from sqlalchemy.orm import selectinload
 from sqlalchemy.sql import select
+from sqlalchemy.sql.expression import cast, column, literal, union
+from sqlalchemy.types import Float
 
 from matcher import database, model, wikidata, wikidata_api
 from matcher.planet import line, point, polygon
@@ -48,11 +51,10 @@ def get_country_iso3166_1(lat: float, lon: float) -> set[str]:
 
     Normally there should be only one country.
     """
-    point = sqlalchemy.func.ST_SetSRID(sqlalchemy.func.ST_MakePoint(lon, lat), srid)
+    point = func.ST_SetSRID(func.ST_MakePoint(lon, lat), srid)
     alpha2_codes: set[str] = set()
     q = model.Polygon.query.filter(
-        sqlalchemy.func.ST_Covers(model.Polygon.way, point),
+        func.ST_Covers(model.Polygon.way, point), model.Polygon.admin_level == "2"
-        model.Polygon.admin_level == "2",
     )
     for country in q:
         alpha2: str = country.tags.get("ISO3166-1")
@@ -88,15 +90,13 @@ def is_street_number_first(lat: float, lon: float) -> bool:
 
 def make_envelope(bounds: list[float]) -> geoalchemy2.functions.ST_MakeEnvelope:
     """Make en envelope for the given bounds."""
-    return sqlalchemy.func.ST_MakeEnvelope(*bounds, srid)
+    return func.ST_MakeEnvelope(*bounds, srid)
 
 
 def get_bbox_centroid(bbox: list[float]) -> tuple[str, str]:
     """Get centroid of bounding box."""
     bbox = make_envelope(bbox)
-    centroid = database.session.query(
+    centroid = database.session.query(func.ST_AsText(func.ST_Centroid(bbox))).scalar()
-        sqlalchemy.func.ST_AsText(sqlalchemy.func.ST_Centroid(bbox))
-    ).scalar()
     m = re_point.match(centroid)
     assert m
     lon, lat = m.groups()
@@ -107,25 +107,16 @@ def get_bbox_centroid(bbox: list[float]) -> tuple[str, str]:
 def make_envelope_around_point(
     lat: float, lon: float, distance: float
 ) -> geoalchemy2.functions.ST_MakeEnvelope:
-    """Make an envelope around a point, the distance parameter specifies the size."""
     conn = database.session.connection()
 
-    p = sqlalchemy.func.ST_MakePoint(lon, lat)
+    p = func.ST_MakePoint(lon, lat)
 
     s = select(
         [
-            sqlalchemy.func.ST_AsText(
+            func.ST_AsText(func.ST_Project(p, distance, func.radians(0))),
-                sqlalchemy.func.ST_Project(p, distance, sqlalchemy.func.radians(0))
+            func.ST_AsText(func.ST_Project(p, distance, func.radians(90))),
-            ),
+            func.ST_AsText(func.ST_Project(p, distance, func.radians(180))),
-            sqlalchemy.func.ST_AsText(
+            func.ST_AsText(func.ST_Project(p, distance, func.radians(270))),
-                sqlalchemy.func.ST_Project(p, distance, sqlalchemy.func.radians(90))
-            ),
-            sqlalchemy.func.ST_AsText(
-                sqlalchemy.func.ST_Project(p, distance, sqlalchemy.func.radians(180))
-            ),
-            sqlalchemy.func.ST_AsText(
-                sqlalchemy.func.ST_Project(p, distance, sqlalchemy.func.radians(270))
-            ),
         ]
     )
     row = conn.execute(s).fetchone()
@@ -136,7 +127,7 @@ def make_envelope_around_point(
     south = coords[2][1]
     west = coords[3][0]
 
-    return sqlalchemy.func.ST_MakeEnvelope(west, south, east, north, srid)
+    return func.ST_MakeEnvelope(west, south, east, north, srid)
 
 
 def drop_way_area(tags: TagsType) -> TagsType:
@@ -155,13 +146,13 @@ def get_part_of(table_name, src_id, bbox):
             [
                 polygon.c.osm_id,
                 polygon.c.tags,
-                sqlalchemy.func.ST_Area(sqlalchemy.func.ST_Collect(polygon.c.way)),
+                func.ST_Area(func.ST_Collect(polygon.c.way)),
             ]
         )
         .where(
             and_(
-                sqlalchemy.func.ST_Intersects(bbox, polygon.c.way),
+                func.ST_Intersects(bbox, polygon.c.way),
-                sqlalchemy.func.ST_Covers(polygon.c.way, table_alias.c.way),
+                func.ST_Covers(polygon.c.way, table_alias.c.way),
                 table_alias.c.osm_id == src_id,
                 polygon.c.tags.has_key("name"),
                 or_(
@@ -236,8 +227,8 @@ def get_items_in_bbox(bbox: list[float]):
 
     q = (
         model.Item.query.join(model.ItemLocation)
-        .filter(sqlalchemy.func.ST_Covers(db_bbox, model.ItemLocation.location))
+        .filter(func.ST_Covers(db_bbox, model.ItemLocation.location))
-        .options(sqlalchemy.orm.selectinload(model.Item.locations))
+        .options(selectinload(model.Item.locations))
     )
 
     return q
@@ -248,7 +239,7 @@ def get_osm_with_wikidata_tag(bbox, isa_filter=None):
     extra_sql = ""
     if isa_filter:
         q = model.Item.query.join(model.ItemLocation).filter(
-            sqlalchemy.func.ST_Covers(make_envelope(bbox), model.ItemLocation.location)
+            func.ST_Covers(make_envelope(bbox), model.ItemLocation.location)
         )
         q = add_isa_filter(q, isa_filter)
         qids = [isa.qid for isa in q]
@@ -283,7 +274,7 @@ WHERE tags ? 'wikidata'
         + extra_sql
     )
     conn = database.session.connection()
-    result = conn.execute(sqlalchemy.text(sql))
+    result = conn.execute(text(sql))
 
     # print(sql)
 
@@ -353,9 +344,7 @@ def get_item_tags(item: model.Item) -> dict[str, list[str]]:
 
     osm_list = collections.defaultdict(list)
 
-    skip_isa: set[int] = {
+    skip_isa = {row[0] for row in database.session.query(model.SkipIsA.item_id)}
-        row[0] for row in database.session.query(model.SkipIsA.item_id)
-    }
 
     tram_stop_id = 41176
     airport_id = 1248784
@@ -363,7 +352,7 @@ def get_item_tags(item: model.Item) -> dict[str, list[str]]:
     if {tram_stop_id, airport_id, aerodrome_id} & set(isa_list):
         skip_isa.add(41176)  # building (Q41176)
 
-    seen: set[int] = set(isa_list) | skip_isa
+    seen = set(isa_list) | skip_isa
     stop = {
         "Q11799049": "public institution",
         "Q7075": "library",
@@ -375,9 +364,7 @@ def get_item_tags(item: model.Item) -> dict[str, list[str]]:
             continue
         isa_qid: str = typing.cast(str, isa.qid)
         isa_path = isa_path + [{"qid": isa_qid, "label": isa.label()}]
-        osm: list[str] = [
+        osm = [v for v in isa.get_claim("P1282") if v not in skip_tags]
-            typing.cast(str, v) for v in isa.get_claim("P1282") if v not in skip_tags
-        ]
 
         osm += [
             extra.tag_or_key
@@ -391,7 +378,7 @@ def get_item_tags(item: model.Item) -> dict[str, list[str]]:
             # item is specific enough, no need to keep walking the item hierarchy
             continue
 
-        check: set[int] = set()
+        check = set()
         properties = [
             ("P279", "subclass of"),
             ("P140", "religion"),
@@ -402,16 +389,12 @@ def get_item_tags(item: model.Item) -> dict[str, list[str]]:
         ]
 
         for pid, label in properties:
-            check |= {
+            check |= {v["numeric-id"] for v in (isa.get_claim(pid) or []) if v}
-                typing.cast(dict[str, int], v)["numeric-id"]
-                for v in (isa.get_claim(pid) or [])
-                if v
-            }
 
         print(isa.qid, isa.label(), check)
-        isa_list_set = check - seen
+        isa_list = check - seen
-        seen.update(isa_list_set)
+        seen.update(isa_list)
-        isa_items += [(isa, isa_path) for isa in get_items(isa_list_set)]
+        isa_items += [(isa, isa_path) for isa in get_items(isa_list)]
     return {key: list(values) for key, values in osm_list.items()}
 
 
@@ -484,7 +467,7 @@ def get_tags_for_isa_item(item):
 
 def add_isa_filter(q, isa_qids):
     q_subclass = database.session.query(model.Item.qid).filter(
-        sqlalchemy.func.jsonb_path_query_array(
+        func.jsonb_path_query_array(
             model.Item.claims,
             "$.P279[*].mainsnak.datavalue.value.id",
         ).bool_op("?|")(list(isa_qids))
@@ -492,7 +475,7 @@ def add_isa_filter(q, isa_qids):
 
     subclass_qid = {qid for qid, in q_subclass.all()}
 
-    isa = sqlalchemy.func.jsonb_path_query_array(
+    isa = func.jsonb_path_query_array(
         model.Item.claims,
         "$.P31[*].mainsnak.datavalue.value.id",
     ).bool_op("?|")
@@ -501,7 +484,7 @@ def add_isa_filter(q, isa_qids):
 
 def wikidata_items_count(bounds, isa_filter=None):
     q = model.Item.query.join(model.ItemLocation).filter(
-        sqlalchemy.func.ST_Covers(make_envelope(bounds), model.ItemLocation.location)
+        func.ST_Covers(make_envelope(bounds), model.ItemLocation.location)
     )
 
     if isa_filter:
@@ -516,7 +499,7 @@ def wikidata_isa_counts(bounds, isa_filter=None):
     db_bbox = make_envelope(bounds)
 
     q = model.Item.query.join(model.ItemLocation).filter(
-        sqlalchemy.func.ST_Covers(db_bbox, model.ItemLocation.location)
+        func.ST_Covers(db_bbox, model.ItemLocation.location)
     )
 
     if isa_filter:
@@ -546,32 +529,25 @@ def wikidata_isa_counts(bounds, isa_filter=None):
     return isa_count
 
 
-def get_tag_filter(
+def get_tag_filter(tags, tag_list):
-    tags: sqlalchemy.sql.schema.Column, tag_list: list[str]
-) -> list[sqlalchemy.sql.elements.BooleanClauseList]:
     tag_filter = []
-
-    include_prefix = len(tag_list) < 10
-
     for tag_or_key in tag_list:
         if tag_or_key.startswith("Key:"):
             key = tag_or_key[4:]
             tag_filter.append(and_(tags.has_key(key), tags[key] != "no"))
-            if include_prefix:
+            for prefix in tag_prefixes:
-                for prefix in tag_prefixes:
+                tag_filter.append(tags.has_key(f"{prefix}:{key}"))
-                    tag_filter.append(tags.has_key(f"{prefix}:{key}"))
 
         if tag_or_key.startswith("Tag:"):
             k, _, v = tag_or_key[4:].partition("=")
             tag_filter.append(tags[k] == v)
-            if include_prefix:
+            for prefix in tag_prefixes:
-                for prefix in tag_prefixes:
+                tag_filter.append(tags[f"{prefix}:{k}"] == v)
-                    tag_filter.append(tags[f"{prefix}:{k}"] == v)
 
     return tag_filter
 
 
-def get_preset_translations() -> dict[str, typing.Any]:
+def get_preset_translations():
     app = flask.current_app
     country_language = {
         "AU": "en-AU",  # Australia
@@ -593,9 +569,7 @@ def get_preset_translations() -> dict[str, typing.Any]:
             continue
 
         try:
-            return typing.cast(
+            return json_data[lang_code]["presets"]["presets"]
-                dict[str, typing.Any], json_data[lang_code]["presets"]["presets"]
-            )
         except KeyError:
             pass
 
@@ -691,13 +665,8 @@ def address_node_label(tags: TagsType) -> str | None:
 def get_address_nodes_within_building(osm_id, bbox_list):
     q = model.Point.query.filter(
         polygon.c.osm_id == osm_id,
-        or_(
+        or_(*[func.ST_Intersects(bbox, model.Point.way) for bbox in bbox_list]),
-            *[
+        func.ST_Covers(polygon.c.way, model.Point.way),
-                sqlalchemy.func.ST_Intersects(bbox, model.Point.way)
-                for bbox in bbox_list
-            ]
-        ),
-        sqlalchemy.func.ST_Covers(polygon.c.way, model.Point.way),
         model.Point.tags.has_key("addr:street"),
         model.Point.tags.has_key("addr:housenumber"),
     )
@@ -739,38 +708,31 @@ def find_osm_candidates(item, limit=80, max_distance=450, names=None):
         for loc in item.locations
     ]
 
-    null_area = sqlalchemy.sql.expression.cast(None, sqlalchemy.types.Float)
+    null_area = cast(None, Float)
-    dist = sqlalchemy.sql.expression.column("dist")
+    dist = column("dist")
-    tags = sqlalchemy.sql.expression.column(
+    tags = column("tags", postgresql.HSTORE)
-        "tags", sqlalchemy.dialects.postgresql.HSTORE
-    )
 
     tag_list = get_item_tags(item)
+    # tag_filters = get_tag_filter(point.c.tags, tag_list)
+    # print(tag_filters)
 
     s_point = (
         select(
             [
-                sqlalchemy.sql.expression.literal("point").label("t"),
+                literal("point").label("t"),
                 point.c.osm_id,
                 point.c.tags.label("tags"),
-                sqlalchemy.func.min(
+                func.min(
-                    sqlalchemy.func.ST_DistanceSphere(
+                    func.ST_DistanceSphere(model.ItemLocation.location, point.c.way)
-                        model.ItemLocation.location, point.c.way
-                    )
                 ).label("dist"),
-                sqlalchemy.func.ST_AsText(point.c.way),
+                func.ST_AsText(point.c.way),
-                sqlalchemy.func.ST_AsGeoJSON(point.c.way),
+                func.ST_AsGeoJSON(point.c.way),
                 null_area,
             ]
         )
         .where(
             and_(
-                or_(
+                or_(*[func.ST_Intersects(bbox, point.c.way) for bbox in bbox_list]),
-                    *[
-                        sqlalchemy.func.ST_Intersects(bbox, point.c.way)
-                        for bbox in bbox_list
-                    ]
-                ),
                 model.ItemLocation.item_id == item_id,
                 or_(*get_tag_filter(point.c.tags, tag_list)),
             )
@@ -781,29 +743,20 @@ def find_osm_candidates(item, limit=80, max_distance=450, names=None):
     s_line = (
         select(
             [
-                sqlalchemy.sql.expression.literal("line").label("t"),
+                literal("line").label("t"),
                 line.c.osm_id,
                 line.c.tags.label("tags"),
-                sqlalchemy.func.min(
+                func.min(
-                    sqlalchemy.func.ST_DistanceSphere(
+                    func.ST_DistanceSphere(model.ItemLocation.location, line.c.way)
-                        model.ItemLocation.location, line.c.way
-                    )
                 ).label("dist"),
-                sqlalchemy.func.ST_AsText(
+                func.ST_AsText(func.ST_Centroid(func.ST_Collect(line.c.way))),
-                    sqlalchemy.func.ST_Centroid(sqlalchemy.func.ST_Collect(line.c.way))
+                func.ST_AsGeoJSON(func.ST_Collect(line.c.way)),
-                ),
-                sqlalchemy.func.ST_AsGeoJSON(sqlalchemy.func.ST_Collect(line.c.way)),
                 null_area,
             ]
         )
         .where(
             and_(
-                or_(
+                or_(*[func.ST_Intersects(bbox, line.c.way) for bbox in bbox_list]),
-                    *[
-                        sqlalchemy.func.ST_Intersects(bbox, line.c.way)
-                        for bbox in bbox_list
-                    ]
-                ),
                 model.ItemLocation.item_id == item_id,
                 or_(*get_tag_filter(line.c.tags, tag_list)),
             )
@@ -814,48 +767,33 @@ def find_osm_candidates(item, limit=80, max_distance=450, names=None):
     s_polygon = (
         select(
             [
-                sqlalchemy.sql.expression.literal("polygon").label("t"),
+                literal("polygon").label("t"),
                 polygon.c.osm_id,
                 polygon.c.tags.label("tags"),
-                sqlalchemy.func.min(
+                func.min(
-                    sqlalchemy.func.ST_DistanceSphere(
+                    func.ST_DistanceSphere(model.ItemLocation.location, polygon.c.way)
-                        model.ItemLocation.location, polygon.c.way
-                    )
                 ).label("dist"),
-                sqlalchemy.func.ST_AsText(
+                func.ST_AsText(func.ST_Centroid(func.ST_Collect(polygon.c.way))),
-                    sqlalchemy.func.ST_Centroid(
+                func.ST_AsGeoJSON(func.ST_Collect(polygon.c.way)),
-                        sqlalchemy.func.ST_Collect(polygon.c.way)
+                func.ST_Area(func.ST_Collect(polygon.c.way)),
-                    )
-                ),
-                sqlalchemy.func.ST_AsGeoJSON(sqlalchemy.func.ST_Collect(polygon.c.way)),
-                sqlalchemy.func.ST_Area(sqlalchemy.func.ST_Collect(polygon.c.way)),
             ]
         )
         .where(
             and_(
-                or_(
+                or_(*[func.ST_Intersects(bbox, polygon.c.way) for bbox in bbox_list]),
-                    *[
-                        sqlalchemy.func.ST_Intersects(bbox, polygon.c.way)
-                        for bbox in bbox_list
-                    ]
-                ),
                 model.ItemLocation.item_id == item_id,
                 or_(*get_tag_filter(polygon.c.tags, tag_list)),
             )
         )
         .group_by(polygon.c.osm_id, polygon.c.tags)
         .having(
-            sqlalchemy.func.ST_Area(sqlalchemy.func.ST_Collect(polygon.c.way))
+            func.ST_Area(func.ST_Collect(polygon.c.way))
-            < 20 * sqlalchemy.func.ST_Area(bbox_list[0])
+            < 20 * func.ST_Area(bbox_list[0])
         )
     )
 
     tables = ([] if item_is_linear_feature else [s_point]) + [s_line, s_polygon]
-    s = (
+    s = select([union(*tables).alias()]).where(dist < max_distance).order_by(dist)
-        select([sqlalchemy.sql.expression.union(*tables).alias()])
-        .where(dist < max_distance)
-        .order_by(dist)
-    )
 
     if names:
         s = s.where(or_(tags["name"].in_(names), tags["old_name"].in_(names)))
@@ -978,27 +916,7 @@ def check_is_street_number_first(latlng):
     flask.g.street_number_first = is_street_number_first(*latlng)
 
 
-class ItemDetailType(typing.TypedDict):
+def item_detail(item):
-    """Details of an item as a dict."""
-
-    qid: str
-    label: str
-    description: str
-    markers: list[dict[str, float]]
-    image_list: list[str]
-    street_address: list[str]
-    isa_list: list[dict[str, str]]
-    closed: bool
-    inception: str
-    p1619: str
-    p576: str
-    heritage_designation: str
-    wikipedia: dict[str, str]
-    identifiers: list[str]
-
-
-def item_detail(item: model.Item) -> ItemDetailType:
-    """Get detail for an item, returns a dict."""
     unsupported_relation_types = {
         "Q194356",  # wind farm
         "Q2175765",  # tram stop
@@ -1135,27 +1053,27 @@ def missing_wikidata_items(qids, lat, lon):
         }
         isa_count.append(isa)
 
-    return {"items": items, "isa_count": isa_count}
+    return dict(items=items, isa_count=isa_count)
 
 
-def isa_incremental_search(search_terms: str) -> list[dict[str, str]]:
+def isa_incremental_search(search_terms):
-    """Incremental search."""
+    en_label = func.jsonb_extract_path_text(model.Item.labels, "en", "value")
-    en_label = sqlalchemy.func.jsonb_extract_path_text(model.Item.labels, "en", "value")
     q = model.Item.query.filter(
         model.Item.claims.has_key("P1282"),
         en_label.ilike(f"%{search_terms}%"),
-        sqlalchemy.func.length(en_label) < 20,
+        func.length(en_label) < 20,
     )
 
     # print(q.statement.compile(compile_kwargs={"literal_binds": True}))
 
-    return [
+    ret = []
-        {
+    for item in q:
+        cur = {
             "qid": item.qid,
             "label": item.label(),
         }
-        for item in q
+        ret.append(cur)
-    ]
+    return ret
 
 
 class PlaceItems(typing.TypedDict):
@@ -1173,7 +1091,7 @@ def get_place_items(osm_type: str, osm_id: int) -> PlaceItems:
         model.Item.query.join(model.ItemLocation)
         .join(
             model.Polygon,
-            sqlalchemy.func.ST_Covers(model.Polygon.way, model.ItemLocation.location),
+            func.ST_Covers(model.Polygon.way, model.ItemLocation.location),
         )
         .filter(model.Polygon.src_id == src_id)
     )

matcher/model.py

@@ -4,7 +4,6 @@ import typing
 from collections import defaultdict
 from typing import Any
 
-import sqlalchemy
 from flask_login import UserMixin
 from geoalchemy2 import Geometry
 from sqlalchemy import func
@@ -16,6 +15,7 @@ from sqlalchemy.orm import backref, column_property, deferred, registry, relatio
 from sqlalchemy.orm.collections import attribute_mapped_collection
 from sqlalchemy.orm.decl_api import DeclarativeMeta
 from sqlalchemy.schema import Column, ForeignKey
+from sqlalchemy.sql.expression import cast
 from sqlalchemy.types import BigInteger, Boolean, DateTime, Float, Integer, String, Text
 
 from . import mail, utils, wikidata
@@ -24,14 +24,7 @@ from .database import now_utc, session
 mapper_registry = registry()
 
 
-def cast_to_string(v: Column[int]) -> sqlalchemy.sql.elements.Cast[str]:
-    """Cast an value to a string."""
-    return sqlalchemy.sql.expression.cast(v, String)
-
-
 class Base(metaclass=DeclarativeMeta):
-    """Database model base class."""
-
     __abstract__ = True
 
     registry = mapper_registry
@@ -101,7 +94,7 @@ class Item(Base):
     locations = relationship(
         "ItemLocation", cascade="all, delete-orphan", backref="item"
     )
-    qid = column_property("Q" + cast_to_string(item_id))
+    qid = column_property("Q" + cast(item_id, String))
 
     wiki_extracts = relationship(
         "Extract",
@@ -113,7 +106,6 @@ class Item(Base):
 
     @classmethod
     def get_by_qid(cls: typing.Type[T], qid: str) -> T | None:
-        """Lookup Item via QID."""
         if qid and len(qid) > 1 and qid[0].upper() == "Q" and qid[1:].isdigit():
             obj: T = cls.query.get(qid[1:])
             return obj
@@ -124,7 +116,7 @@ class Item(Base):
         """Wikidata URL for item."""
         return f"https://www.wikidata.org/wiki/{self.qid}"
 
-    def get_claim(self, pid: str) -> list[str | int | dict[str, str | int] | None]:
+    def get_claim(self, pid: str) -> list[dict[str, Any] | None]:
         """List of claims for given Wikidata property ID."""
         claims = typing.cast(wikidata.Claims, self.claims)
         return [
@@ -254,9 +246,8 @@ class Item(Base):
                     isa_list.append(of_qualifier["datavalue"]["value"])
         return isa_list
 
-    def get_isa_qids(self) -> list[str]:
+    def get_isa_qids(self):
-        """Get QIDs of items listed instance of (P31) property."""
+        return [isa["id"] for isa in self.get_isa()]
-        return [typing.cast(str, isa["id"]) for isa in self.get_isa()]
 
     def is_street(self, isa_qids=None):
         if isa_qids is None:
@@ -290,12 +281,10 @@ class Item(Base):
         isa_qids = set(self.get_isa_qids())
         return self.is_street(isa_qids) or self.is_watercourse(isa_qids)
 
-    def is_tram_stop(self) -> bool:
+    def is_tram_stop(self):
-        """Item is a tram stop."""
         return "Q2175765" in self.get_isa_qids()
 
-    def alert_admin_about_bad_time(self, v: utils.WikibaseTime) -> None:
+    def alert_admin_about_bad_time(self, v):
-        """Send an email to admin when encountering an unparsable time in Wikibase."""
         body = (
             "Wikidata item has an unsupported time precision\n\n"
             + self.wd_url
@@ -305,10 +294,9 @@ class Item(Base):
         )
         mail.send_mail(f"OWL Map: bad time value in {self.qid}", body)
 
-    def time_claim(self, pid: str) -> list[str]:
+    def time_claim(self, pid):
-        """Read values from time statement."""
         ret = []
-        for v in typing.cast(list[utils.WikibaseTime | None], self.get_claim(pid)):
+        for v in self.get_claim(pid):
             if not v:
                 continue
             try:
@@ -324,18 +312,15 @@ class Item(Base):
 
         return ret
 
-    def closed(self) -> list[str]:
+    def closed(self):
-        """Date when item closed."""
         return self.time_claim("P3999")
 
-    def first_paragraph_language(self, lang: str) -> str | None:
+    def first_paragraph_language(self, lang):
-        """First paragraph of Wikipedia article in the given languages."""
         if lang not in self.sitelinks():
-            return None
+            return
         extract = self.extracts.get(lang)
         if not extract:
-            return None
+            return
-        assert isinstance(extract, str)
 
         empty_list = [
             "<p><span></span></p>",
@@ -414,8 +399,6 @@ class Item(Base):
 
 
 class ItemIsA(Base):
-    """Item IsA."""
-
     __tablename__ = "item_isa"
     item_id = Column(Integer, ForeignKey("item.item_id"), primary_key=True)
     isa_id = Column(Integer, ForeignKey("item.item_id"), primary_key=True)
@@ -425,16 +408,14 @@ class ItemIsA(Base):
 
 
 class ItemLocation(Base):
-    """Location of an item."""
-
     __tablename__ = "item_location"
     item_id = Column(Integer, ForeignKey("item.item_id"), primary_key=True)
     property_id = Column(Integer, primary_key=True)
     statement_order = Column(Integer, primary_key=True)
     location = Column(Geometry("POINT", srid=4326, spatial_index=True), nullable=False)
 
-    qid = column_property("Q" + cast_to_string(item_id))
+    qid = column_property("Q" + cast(item_id, String))
-    pid = column_property("P" + cast_to_string(property_id))
+    pid = column_property("P" + cast(item_id, String))
 
     def get_lat_lon(self) -> tuple[float, float]:
         """Get latitude and longitude of item."""
@@ -652,7 +633,7 @@ class SkipIsA(Base):
 
     __tablename__ = "skip_isa"
     item_id = Column(Integer, ForeignKey("item.item_id"), primary_key=True)
-    qid = column_property("Q" + cast_to_string(item_id))
+    qid = column_property("Q" + cast(item_id, String))
 
     item = relationship("Item")
 
@@ -664,7 +645,7 @@ class ItemExtraKeys(Base):
     item_id = Column(Integer, ForeignKey("item.item_id"), primary_key=True)
     tag_or_key = Column(String, primary_key=True)
     note = Column(String)
-    qid = column_property("Q" + cast_to_string(item_id))
+    qid = column_property("Q" + cast(item_id, String))
 
     item = relationship("Item")
 

notes

@@ -404,11 +404,3 @@ select item_id, labels->'en'->'value'
 ---
 Shipyard results shouldn't include place=city
 https://map.osm.wikidata.link/map/18/50.89540/-1.38243?item=Q551401
-
----
-2023-05-19
-
-Need option to show labels in more languages. Matching in Greece doesn't work
-great when there is an English label in Wikidata. OWL Map shows the English
-label, if the object in OSM only has a label in Greek then it is hard to tell if
-they match. Should optionally show more languages.

web_view.py

@@ -1,7 +1,5 @@
 #!/usr/bin/python3
 
-"""Views for the web app."""
-
 import json
 import re
 from time import sleep, time
@@ -297,7 +295,7 @@ def search_page() -> str:
 
 
 @app.route("/map/<int:zoom>/<float(signed=True):lat>/<float(signed=True):lon>")
-def map_location(zoom: int, lat: float, lon: float) -> str:
+def map_location(zoom, lat, lon):
     qid = flask.request.args.get("item")
     isa_param = flask.request.args.get("isa")
     if qid: