(sql_query: str)
| 21 | |
| 22 | |
| 23 | def execute_sql(sql_query: str): |
| 24 | if not is_read_only_query(sql_query): |
| 25 | raise NotReadOnlyException("Only read-only queries are allowed.") |
| 26 | |
| 27 | with ENGINE.connect() as connection: |
| 28 | connection = connection.execution_options( |
| 29 | postgresql_readonly=True |
| 30 | ) |
| 31 | with connection.begin(): |
| 32 | sql_text = text(sql_query) |
| 33 | result = connection.execute(sql_text) |
| 34 | |
| 35 | column_names = list(result.keys()) |
| 36 | if 'state' not in column_names and any(c in column_names for c in ['city', 'county']): |
| 37 | raise CityOrCountyWithoutStateException("Include `state` in the result table, too.") |
| 38 | |
| 39 | rows = [list(r) for r in result.all()] |
| 40 | |
| 41 | # Add lat and lon to zip_code |
| 42 | zip_code_idx = None |
| 43 | try: |
| 44 | zip_code_idx = column_names.index("zip_code") |
| 45 | except ValueError: |
| 46 | zip_code_idx = None |
| 47 | |
| 48 | if zip_code_idx is not None: |
| 49 | column_names.append("lat") |
| 50 | column_names.append("long") |
| 51 | for row in rows: |
| 52 | zip_code = row[zip_code_idx] |
| 53 | lat = zip_lat_lon.get(zip_code, {}).get('lat') |
| 54 | lon = zip_lat_lon.get(zip_code, {}).get('lon') |
| 55 | row.append(lat) |
| 56 | row.append(lon) |
| 57 | |
| 58 | # No zip_code lat lon, so try to get city lat lon |
| 59 | else: |
| 60 | # Add lat and lon to city |
| 61 | city_idx = None |
| 62 | state_idx = None |
| 63 | try: |
| 64 | city_idx = column_names.index("city") |
| 65 | state_idx = column_names.index("state") |
| 66 | except ValueError: |
| 67 | city_idx = None |
| 68 | state_idx = None |
| 69 | |
| 70 | if city_idx is not None and state_idx is not None: |
| 71 | column_names.append("lat") |
| 72 | column_names.append("long") |
| 73 | for row in rows: |
| 74 | city = row[city_idx] |
| 75 | state = row[state_idx] |
| 76 | lat = city_lat_lon.get(state, {}).get(city, {}).get('lat') |
| 77 | lon = city_lat_lon.get(state, {}).get(city, {}).get('lon') |
| 78 | |
| 79 | if "St." in city: |
| 80 | new_city = city.replace("St.", "Saint") |
no test coverage detected