Skip to content

Commit

Permalink
Fix typing in geoname views and models
Browse files Browse the repository at this point in the history
  • Loading branch information
jace committed Dec 15, 2023
1 parent 183a099 commit 609be70
Show file tree
Hide file tree
Showing 3 changed files with 43 additions and 40 deletions.
51 changes: 29 additions & 22 deletions funnel/models/geoname.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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."""

Expand Down Expand Up @@ -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:
Expand All @@ -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),
Expand All @@ -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)

Expand All @@ -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),
Expand All @@ -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),
Expand All @@ -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)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand All @@ -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]
Expand Down Expand Up @@ -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))
Expand All @@ -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,
Expand Down
18 changes: 9 additions & 9 deletions funnel/views/api/geoname.py
Original file line number Diff line number Diff line change
Expand Up @@ -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))
Expand All @@ -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:
Expand All @@ -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',
Expand All @@ -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)
Expand All @@ -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',
Expand Down
14 changes: 5 additions & 9 deletions funnel/views/jobs.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down

0 comments on commit 609be70

Please sign in to comment.