From 609be703a2b58f30282f30344affbada956d2a9b Mon Sep 17 00:00:00 2001 From: Kiran Jonnalagadda Date: Fri, 15 Dec 2023 23:31:47 +0530 Subject: [PATCH] Fix typing in geoname views and models --- funnel/models/geoname.py | 51 +++++++++++++++++++++---------------- funnel/views/api/geoname.py | 18 ++++++------- funnel/views/jobs.py | 14 ++++------ 3 files changed, 43 insertions(+), 40 deletions(-) diff --git a/funnel/models/geoname.py b/funnel/models/geoname.py index f63d48ebb..3261e996a 100644 --- a/funnel/models/geoname.py +++ b/funnel/models/geoname.py @@ -6,7 +6,7 @@ from collections.abc import Collection from datetime import date from decimal import Decimal -from typing import Self, cast +from typing import Required, Self, TypedDict from sqlalchemy.dialects.postgresql import ARRAY @@ -43,6 +43,12 @@ } +class ParseLocationsDict(TypedDict, total=False): + token: Required[str] + special: bool + geoname: GeoName + + class GeoCountryInfo(BaseNameMixin, GeonameModel): """Geoname record for a country.""" @@ -109,7 +115,7 @@ class GeoAdmin1Code(BaseMixin, GeonameModel): country_id: Mapped[str | None] = sa_orm.mapped_column( 'country', sa.CHAR(2), sa.ForeignKey('geo_country_info.iso_alpha2') ) - country: Mapped[GeoCountryInfo | None] = relationship('GeoCountryInfo') + country: Mapped[GeoCountryInfo | None] = relationship() admin1_code: Mapped[str | None] = sa_orm.mapped_column(sa.Unicode) def __repr__(self) -> str: @@ -124,7 +130,6 @@ class GeoAdmin2Code(BaseMixin, GeonameModel): geonameid: Mapped[int] = sa_orm.synonym('id') geoname: Mapped[GeoName] = relationship( - 'GeoName', uselist=False, viewonly=True, primaryjoin=lambda: GeoAdmin2Code.id == sa_orm.foreign(GeoName.id), @@ -135,7 +140,7 @@ class GeoAdmin2Code(BaseMixin, GeonameModel): country_id: Mapped[str | None] = sa_orm.mapped_column( 'country', sa.CHAR(2), sa.ForeignKey('geo_country_info.iso_alpha2') ) - country: Mapped[GeoCountryInfo | None] = relationship('GeoCountryInfo') + country: Mapped[GeoCountryInfo | None] = relationship() admin1_code: Mapped[str | None] = sa_orm.mapped_column(sa.Unicode) admin2_code: Mapped[str | None] = sa_orm.mapped_column(sa.Unicode) @@ -158,11 +163,10 @@ class GeoName(BaseNameMixin, GeonameModel): country_id: Mapped[str | None] = sa_orm.mapped_column( 'country', sa.CHAR(2), sa.ForeignKey('geo_country_info.iso_alpha2') ) - country: Mapped[GeoCountryInfo | None] = relationship('GeoCountryInfo') + country: Mapped[GeoCountryInfo | None] = relationship() cc2: Mapped[str | None] = sa_orm.mapped_column(sa.Unicode) admin1: Mapped[str | None] = sa_orm.mapped_column(sa.Unicode) admin1_ref: Mapped[GeoAdmin1Code | None] = relationship( - 'GeoAdmin1Code', uselist=False, primaryjoin=lambda: sa.and_( GeoName.country_id == sa_orm.foreign(GeoAdmin1Code.country_id), @@ -174,12 +178,11 @@ class GeoName(BaseNameMixin, GeonameModel): sa.Integer, sa.ForeignKey('geo_admin1_code.id'), nullable=True ) admin1code: Mapped[GeoAdmin1Code | None] = relationship( - 'GeoAdmin1Code', uselist=False, foreign_keys=[admin1_id] + uselist=False, foreign_keys=[admin1_id] ) admin2: Mapped[str | None] = sa_orm.mapped_column(sa.Unicode) admin2_ref: Mapped[GeoAdmin2Code | None] = relationship( - 'GeoAdmin2Code', uselist=False, primaryjoin=lambda: sa.and_( GeoName.country_id == sa_orm.foreign(GeoAdmin2Code.country_id), @@ -192,7 +195,7 @@ class GeoName(BaseNameMixin, GeonameModel): sa.Integer, sa.ForeignKey('geo_admin2_code.id'), nullable=True ) admin2code: Mapped[GeoAdmin2Code | None] = relationship( - 'GeoAdmin2Code', uselist=False, foreign_keys=[admin2_id] + uselist=False, foreign_keys=[admin2_id] ) admin4: Mapped[str | None] = sa_orm.mapped_column(sa.Unicode) @@ -362,7 +365,11 @@ def related_geonames(self) -> dict[str, GeoName]: related: dict[str, GeoName] = {} if self.admin2code and self.admin2code.geonameid != self.geonameid: related['admin2'] = self.admin2code.geoname - if self.admin1code and self.admin1code.geonameid != self.geonameid: + if ( + self.admin1code + and self.admin1code.geonameid != self.geonameid + and self.admin1code.geoname is not None + ): related['admin1'] = self.admin1code.geoname if ( self.country @@ -473,7 +480,7 @@ def parse_locations( special: list[str] | None = None, lang: str | None = None, bias: list[str] | None = None, - ): + ) -> list[ParseLocationsDict]: """ Parse a string and return annotations marking all identified locations. @@ -485,12 +492,15 @@ def parse_locations( special = [s.lower() for s in special] if special else [] if bias is None: bias = [] - tokens = NOWORDS_RE.split(q) + while '' in bias: + bias.remove('') + bias = [each.upper() for each in bias] + tokens: list[str] = NOWORDS_RE.split(q) while '' in tokens: tokens.remove('') # Remove blank tokens from beginning and end ltokens = [t.lower() for t in tokens] - results: list[dict[str, object]] = [] - counter = 0 + results: list[ParseLocationsDict] = [] + counter: int = 0 limit = len(tokens) while counter < limit: token = tokens[counter] @@ -549,7 +559,7 @@ def parse_locations( candidates = [ (NOWORDS_RE.split(m.title.lower()), m) for m in matches ] - fullmatch = [] + fullmatch: list[tuple[int, GeoAltName]] = [] for mtokens, match in candidates: if mtokens == ltokens[counter : counter + len(mtokens)]: fullmatch.append((len(mtokens), match)) @@ -561,14 +571,11 @@ def parse_locations( # (d) population accepted.sort( key=lambda a: ( - { - v: k - for k, v in enumerate( - reversed(cast(list[str], bias)) - ) - }.get(a.geoname.country_id, -1), + {v: k for k, v in enumerate(reversed(bias))}.get( + a.geoname.country_id or '', -1 + ), {lang: 0}.get(a.lang, 1), - {'A': 1, 'P': 2}.get(a.geoname.fclass, 0), + {'A': 1, 'P': 2}.get(a.geoname.fclass or '', 0), a.geoname.population, ), reverse=True, diff --git a/funnel/views/api/geoname.py b/funnel/views/api/geoname.py index 00c90c463..dc993a497 100644 --- a/funnel/views/api/geoname.py +++ b/funnel/views/api/geoname.py @@ -2,19 +2,21 @@ from __future__ import annotations +from typing import cast + from coaster.utils import getbool from coaster.views import requestargs from ... import app from ...models import GeoName -from ...typing import ReturnRenderWith +from ...typing import ReturnView @app.route('/api/1/geo/get_by_name') @requestargs('name', ('related', getbool), ('alternate_titles', getbool)) def geo_get_by_name( name: str, related: bool = False, alternate_titles: bool = False -) -> ReturnRenderWith: +) -> ReturnView: """Get a geoname record given a single URL stub name or geoname id.""" if name.isdigit(): geoname = GeoName.query.get(int(name)) @@ -36,7 +38,7 @@ def geo_get_by_name( @requestargs('name[]', ('related', getbool), ('alternate_titles', getbool)) def geo_get_by_names( name: list[str], related: bool = False, alternate_titles: bool = False -) -> ReturnRenderWith: +) -> ReturnView: """Get geoname records matching given URL stub names or geonameids.""" geonames = [] for n in name: @@ -57,7 +59,7 @@ def geo_get_by_names( @app.route('/api/1/geo/get_by_title') @requestargs('title[]', 'lang') -def geo_get_by_title(title: list[str], lang: str | None = None) -> ReturnRenderWith: +def geo_get_by_title(title: list[str], lang: str | None = None) -> ReturnView: """Get locations matching given titles.""" return { 'status': 'ok', @@ -73,9 +75,9 @@ def geo_parse_location( lang: str | None = None, bias: list[str] | None = None, alternate_titles: bool = False, -) -> ReturnRenderWith: +) -> ReturnView: """Parse locations from a string of locations.""" - result = GeoName.parse_locations(q, special, lang, bias) + result = cast(list[dict], GeoName.parse_locations(q, special, lang, bias)) for item in result: if 'geoname' in item: item['geoname'] = item['geoname'].as_dict(alternate_titles=alternate_titles) @@ -84,9 +86,7 @@ def geo_parse_location( @app.route('/api/1/geo/autocomplete') @requestargs('q', 'lang', ('limit', int)) -def geo_autocomplete( - q: str, lang: str | None = None, limit: int = 100 -) -> ReturnRenderWith: +def geo_autocomplete(q: str, lang: str | None = None, limit: int = 100) -> ReturnView: """Autocomplete a geoname record.""" return { 'status': 'ok', diff --git a/funnel/views/jobs.py b/funnel/views/jobs.py index 6282f1821..4a6f67d0f 100644 --- a/funnel/views/jobs.py +++ b/funnel/views/jobs.py @@ -78,21 +78,17 @@ def import_tickets(ticket_client_id: int) -> None: @rqjob() def tag_locations(project_id: int) -> None: - """ - Tag a project with geoname locations. - - This function used to retrieve data from Hascore, which has been merged into Funnel - and is available directly as the GeoName model. This code continues to operate with - the legacy Hascore data structure, and is pending rewrite. - """ + """Tag a project with geoname locations. This is legacy code pending a rewrite.""" project = Project.query.get(project_id) + if project is None: + return if not project.location: return results = GeoName.parse_locations( project.location, special=["Internet", "Online"], bias=['IN', 'US'] ) - geonames = defaultdict(dict) - tokens = [] + geonames: dict[str, dict] = defaultdict(dict) + tokens: list[dict] = [] for item in results: if 'geoname' in item: geoname = item['geoname'].as_dict(alternate_titles=False)