update typing for latest changes in SQLAlchemy

This commit is contained in:
Sarah Hoffmann
2023-12-29 20:55:33 +01:00
parent af85ad390f
commit 93afe5a7c3
3 changed files with 7 additions and 7 deletions

View File

@@ -590,7 +590,7 @@ async def complete_address_details(conn: SearchConnection, results: List[BaseRes
return return
ltab = sa.func.JsonArrayEach(sa.type_coerce(lookup_ids, sa.JSON))\ ltab = sa.func.JsonArrayEach(sa.type_coerce(lookup_ids, sa.JSON))\
.table_valued(sa.column('value', type_=sa.JSON)) # type: ignore[no-untyped-call] .table_valued(sa.column('value', type_=sa.JSON))
t = conn.t.placex t = conn.t.placex
taddr = conn.t.addressline taddr = conn.t.addressline
@@ -653,7 +653,7 @@ async def complete_address_details(conn: SearchConnection, results: List[BaseRes
parent_lookup_ids = list(filter(lambda e: e['pid'] != e['lid'], lookup_ids)) parent_lookup_ids = list(filter(lambda e: e['pid'] != e['lid'], lookup_ids))
if parent_lookup_ids: if parent_lookup_ids:
ltab = sa.func.JsonArrayEach(sa.type_coerce(parent_lookup_ids, sa.JSON))\ ltab = sa.func.JsonArrayEach(sa.type_coerce(parent_lookup_ids, sa.JSON))\
.table_valued(sa.column('value', type_=sa.JSON)) # type: ignore[no-untyped-call] .table_valued(sa.column('value', type_=sa.JSON))
sql = sa.select(ltab.c.value['pid'].as_integer().label('src_place_id'), sql = sa.select(ltab.c.value['pid'].as_integer().label('src_place_id'),
t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name, t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
t.c.class_, t.c.type, t.c.extratags, t.c.class_, t.c.type, t.c.extratags,

View File

@@ -7,7 +7,7 @@
""" """
Implementation of the acutal database accesses for forward search. Implementation of the acutal database accesses for forward search.
""" """
from typing import List, Tuple, AsyncIterator, Dict, Any, Callable from typing import List, Tuple, AsyncIterator, Dict, Any, Callable, cast
import abc import abc
import sqlalchemy as sa import sqlalchemy as sa
@@ -190,7 +190,7 @@ def _int_list_to_subquery(inp: List[int]) -> 'sa.Subquery':
as rows in the column 'nr'. as rows in the column 'nr'.
""" """
vtab = sa.func.JsonArrayEach(sa.type_coerce(inp, sa.JSON))\ vtab = sa.func.JsonArrayEach(sa.type_coerce(inp, sa.JSON))\
.table_valued(sa.column('value', type_=sa.JSON)) # type: ignore[no-untyped-call] .table_valued(sa.column('value', type_=sa.JSON))
return sa.select(sa.cast(sa.cast(vtab.c.value, sa.Text), sa.Integer).label('nr')).subquery() return sa.select(sa.cast(sa.cast(vtab.c.value, sa.Text), sa.Integer).label('nr')).subquery()
@@ -656,7 +656,7 @@ class PlaceSearch(AbstractSearch):
.where(tpc.c.postcode.in_(pcs))\ .where(tpc.c.postcode.in_(pcs))\
.scalar_subquery() .scalar_subquery()
penalty += sa.case((t.c.postcode.in_(pcs), 0.0), penalty += sa.case((t.c.postcode.in_(pcs), 0.0),
else_=sa.func.coalesce(pc_near, 2.0)) else_=sa.func.coalesce(pc_near, cast(SaColumn, 2.0)))
if details.viewbox is not None: if details.viewbox is not None:
if details.bounded_viewbox: if details.bounded_viewbox:

View File

@@ -55,7 +55,7 @@ class IntersectsReverseDistance(sa.sql.functions.GenericFunction[Any]):
inherit_cache = True inherit_cache = True
def __init__(self, table: sa.Table, geom: SaColumn) -> None: def __init__(self, table: sa.Table, geom: SaColumn) -> None:
super().__init__(table.c.geometry, # type: ignore[no-untyped-call] super().__init__(table.c.geometry,
table.c.rank_search, geom) table.c.rank_search, geom)
self.tablename = table.name self.tablename = table.name
@@ -122,7 +122,7 @@ class IsAddressPoint(sa.sql.functions.GenericFunction[Any]):
inherit_cache = True inherit_cache = True
def __init__(self, table: sa.Table) -> None: def __init__(self, table: sa.Table) -> None:
super().__init__(table.c.rank_address, # type: ignore[no-untyped-call] super().__init__(table.c.rank_address,
table.c.housenumber, table.c.name) table.c.housenumber, table.c.name)