veekun_pokedex/pokedex/search.py
Andrew Ekstedt 98696cdd69 Fix pokedex search command
Filtering on pokemon name was broken. Not sure why, since the search CLI
was added way after the i18n stuff was added. The error is related to
AssociationProxy, which figures because nothing about association
proxies ever seems to work right. I don't know enough about SQLAlchemy
internals to know if what it was trying to do was supposed to
work, or how to fix it if so. So, fix it by using the same boring
join-based filtering that spline-pokedex uses.

Running `pokedex search --name=gloom` caused the following error with
PostgreSQL and SQLAlchemy 1.3.5:

    sqlalchemy.exc.ProgrammingError: (psycopg2.ProgrammingError) can't adapt type 'ColumnAssociationProxyInstance'

...and this error with SQLite.

    sqlalchemy.exc.InterfaceError: (sqlite3.InterfaceError) Error binding parameter 4 - probably unsupported type.
    [parameters: (9, 9, 9, 9, ColumnAssociationProxyInstance(AssociationProxy('names_local', 'name')), 'gloom')]

Additionaly, the following error would happen with PostgreSQL and SQLAlchemy 0.9.7,
but that's probably unrelated:

    sqlalchemy.exc.ProgrammingError: (ProgrammingError) subquery in FROM must have an alias
    LINE 4: FROM pokemon_species, (SELECT pokemon_species_names.name AS ...
                                   ^

Fixes #296
2020-04-17 16:42:47 -07:00

68 lines
1.7 KiB
Python

import re
from sqlalchemy import func
from sqlalchemy.orm import joinedload
import pokedex.db.tables as t
def _parse_range(value):
v = int(value)
return lambda x: x == v
CRITERION_RX = re.compile(r"""
\s*
(?: (?P<field>[-_a-zA-Z0-9]+): )?
(?P<pattern>
(?:
[^\s"]+?
)+
)
""", re.VERBOSE)
def parse_search_string(string):
"""Parses a search string!"""
criteria = {}
for match in CRITERION_RX.finditer(string):
# TODO what if there are several of the same match!
# TODO the cli needs to do append too
field = match.group('field') or '*'
criteria[field] = match.group('pattern')
return criteria
def search(session, **criteria):
query = (
session.query(t.Pokemon)
.options(
joinedload(t.Pokemon.species)
)
)
stat_query = (
session.query(t.PokemonStat.pokemon_id)
.join(t.PokemonStat.stat)
)
do_stat = False
if criteria.get('name') is not None:
query = query.join(t.Pokemon.species)
query = query.join(t.PokemonSpecies.names_local)
query = query.filter(func.lower(t.PokemonSpecies.names_table.name) == criteria['name'].lower())
for stat_ident in (u'attack', u'defense', u'special-attack', u'special-defense', u'speed', u'hp'):
criterion = criteria.get(stat_ident)
if criterion is None:
continue
do_stat = True
stat_query = stat_query.filter(
(t.Stat.identifier == stat_ident)
& _parse_range(criterion)(t.PokemonStat.base_stat)
)
if do_stat:
query = query.filter(t.Pokemon.id.in_(stat_query.subquery()))
return query.all()