update dependencies

findthatpostcode/__init__.py

@@ -1 +1,3 @@
 from .main import app
+
+__all__ = ["app"]

findthatpostcode/api/__init__.py

@@ -1,12 +1,9 @@
 import logging
-from typing import List
 
-from botocore.client import BaseClient
 from elasticsearch import Elasticsearch
 from fastapi import APIRouter, Depends, Form, HTTPException, Query, status
 from mypy_boto3_s3.client import S3Client
 from pydantic_geojson import FeatureModel
-from sqlalchemy.orm import Session
 
 from findthatpostcode import crud
 from findthatpostcode.db import get_db, get_s3_client

findthatpostcode/commands/boundaries.py

@@ -32,7 +32,6 @@ def import_boundaries(
     es_index: str = AREA_INDEX,
     remove: bool = False,
 ):
-
     es = db.get_db()
 
     if remove:

findthatpostcode/commands/codes.py

@@ -3,7 +3,6 @@
 """
 import codecs
 import csv
-import datetime
 import io
 import zipfile
 from collections import defaultdict
@@ -33,7 +32,6 @@
 @click.option("--es-index", default=ENTITY_INDEX)
 @click.option("--file", default=None)
 def import_rgc(url=settings.RGC_URL, es_index=ENTITY_INDEX, file=None):
-
     if settings.DEBUG:
         requests_cache.install_cache()
 
@@ -75,7 +73,6 @@ def import_chd(
     file=None,
     encoding=settings.DEFAULT_ENCODING,
 ):
-
     if settings.DEBUG:
         requests_cache.install_cache()
 
@@ -91,9 +88,9 @@ def import_chd(
     areas_cache = defaultdict(list)
     areas = {}
 
-    change_history = None
-    changes = None
-    equivalents = None
+    change_history: str | None = None
+    changes: str | None = None
+    equivalents: str | None = None
     for f in z.namelist():
         if f.lower().startswith("changehistory") and f.lower().endswith(".csv"):
             change_history = f
@@ -102,6 +99,13 @@ def import_chd(
         elif f.lower().startswith("equivalents") and f.lower().endswith(".csv"):
             equivalents = f
 
+    if change_history is None:
+        raise Exception("Change history file not found")
+    if changes is None:
+        raise Exception("Changes file not found")
+    if equivalents is None:
+        raise Exception("Equivalents file not found")
+
     with z.open(change_history, "r") as infile:
         click.echo("Opening {}".format(infile.name))
         reader = csv.DictReader(io.TextIOWrapper(infile, encoding))
@@ -195,7 +199,6 @@ def import_chd(
 def import_msoa_names(
     url=settings.MSOA_URL, es_index=AREA_INDEX, file=None, encoding="utf-8-sig"
 ):
-
     if settings.DEBUG:
         requests_cache.install_cache()
 

findthatpostcode/commands/placenames.py

@@ -72,7 +72,6 @@
 def import_placenames(
     url=settings.PLACENAMES_URL, es_index=PLACENAMES_INDEX, file=None
 ):
-
     if settings.DEBUG:
         requests_cache.install_cache()
 

findthatpostcode/commands/postcodes.py

@@ -22,7 +22,6 @@
 @click.option("--url", default=settings.NSPL_URL)
 @click.option("--file", default=None)
 def import_nspl(url=settings.NSPL_URL, es_index=PC_INDEX, file=None):
-
     if settings.DEBUG:
         requests_cache.install_cache()
 

findthatpostcode/commands/stats.py

@@ -8,7 +8,6 @@
 import requests
 import requests_cache
 import tqdm
-from elasticsearch.helpers import bulk
 
 from findthatpostcode import db, settings
 from findthatpostcode.documents import Area
@@ -100,7 +99,6 @@ def parse_field(k, v):
 @click.option("--es-index", default=AREA_INDEX)
 @click.option("--url", default=settings.IMD2019_URL)
 def import_imd2019(url=settings.IMD2019_URL, es_index=AREA_INDEX):
-
     if settings.DEBUG:
         requests_cache.install_cache()
 
@@ -139,14 +137,13 @@ def import_imd2019(url=settings.IMD2019_URL, es_index=AREA_INDEX):
                     }
                 },
             }
-        importer.add(area_update)
+            importer.add(area_update)
 
 
 @click.command("imd2015")
 @click.option("--es-index", default=AREA_INDEX)
 @click.option("--url", default=settings.IMD2015_URL)
 def import_imd2015(url=settings.IMD2015_URL, es_index=AREA_INDEX):
-
     if settings.DEBUG:
         requests_cache.install_cache()
 
@@ -155,7 +152,6 @@ def import_imd2015(url=settings.IMD2015_URL, es_index=AREA_INDEX):
     r = requests.get(url, stream=True)
 
     reader = csv.DictReader(codecs.iterdecode(r.iter_lines(), "utf-8-sig"))
-    area_updates = []
     with BulkImporter(es, "imd2015") as importer:
         for k, area in tqdm.tqdm(enumerate(reader)):
             area = {

findthatpostcode/crud.py

@@ -7,27 +7,19 @@
     Dict,
     Generator,
     List,
-    Literal,
     Optional,
     Tuple,
     Type,
-    TypeVar,
-    Union,
     overload,
 )
 
-from botocore.client import BaseClient
-from botocore.exceptions import ClientError
 from elasticsearch import Elasticsearch
-from elasticsearch_dsl import Document, Q
-from geoalchemy2.comparator import Comparator
+from elasticsearch_dsl import Q
 from mypy_boto3_s3.client import S3Client
-from pydantic.dataclasses import dataclass
 from pydantic_geojson import FeatureModel
 
 from findthatpostcode import schemas, settings
-from findthatpostcode.db import get_db
-from findthatpostcode.documents import Area, Entity, Placename, Postcode
+from findthatpostcode.documents import Area, Placename, Postcode
 from findthatpostcode.utils import PostcodeStr
 
 logger = logging.getLogger(__name__)
@@ -66,7 +58,7 @@ def record_to_schema(
     schema: Type[schemas.Area],
     name_fields: Optional[List[str]] = None,
     name_lookup: Optional[Dict[str, Optional[str]]] = None,
-    **kwargs
+    **kwargs,
 ) -> schemas.Area:
     ...
 
@@ -77,7 +69,7 @@ def record_to_schema(
     schema: Type[schemas.Placename],
     name_fields: Optional[List[str]] = None,
     name_lookup: Optional[Dict[str, Optional[str]]] = None,
-    **kwargs
+    **kwargs,
 ) -> schemas.Placename:
     ...
 
@@ -88,7 +80,7 @@ def record_to_schema(
     schema: Type[schemas.NearestPoint],
     name_fields: Optional[List[str]] = None,
     name_lookup: Optional[Dict[str, Optional[str]]] = None,
-    **kwargs
+    **kwargs,
 ) -> schemas.NearestPoint:
     ...
 
@@ -99,7 +91,7 @@ def record_to_schema(
     schema: Type[schemas.Postcode],
     name_fields: Optional[List[str]] = None,
     name_lookup: Optional[Dict[str, Optional[str]]] = None,
-    **kwargs
+    **kwargs,
 ) -> schemas.Postcode:
     ...
 
@@ -109,12 +101,12 @@ def record_to_schema(
     schema,
     name_fields: Optional[List[str]] = None,
     name_lookup: Optional[Dict[str, Optional[str]]] = None,
-    **kwargs
+    **kwargs,
 ):
     schema_keys = {field.name for field in dataclasses.fields(schema)}
     record_values = dict(
         **kwargs,
-        **{k: v for k, v in record.__dict__["_d_"].items() if k in schema_keys}
+        **{k: v for k, v in record.__dict__["_d_"].items() if k in schema_keys},
     )
     new_record = schema(**record_values)
     if name_fields and name_lookup:
@@ -338,17 +330,17 @@ def search_areas(
         result = db.search(
             index="geo_area,geo_placename",
             body=query,
-            from_=pagination.from_,
-            size=pagination.size,
-            _source_excludes=["boundary"],
-            ignore=[404],
+            from_=pagination.from_,  # type: ignore
+            size=pagination.size,  # type: ignore
+            _source_excludes=["boundary"],  # type: ignore
+            ignore=[404],  # type: ignore
         )
     else:
         result = db.search(
             index="geo_area,geo_placename",
             body=query,
-            _source_excludes=["boundary"],
-            ignore=[404],
+            _source_excludes=["boundary"],  # type: ignore
+            ignore=[404],  # type: ignore
         )
     return_result = []
     for a in result.get("hits", {}).get("hits", []):

findthatpostcode/db.py

@@ -1,7 +1,7 @@
 import click
 from boto3 import session
 from elasticsearch import Elasticsearch
-from elasticsearch_dsl import Document, Index
+from elasticsearch_dsl import Index
 from mypy_boto3_s3.client import S3Client
 
 from findthatpostcode import documents, settings

findthatpostcode/documents/entity.py

@@ -31,7 +31,6 @@ class Index:
 
     @classmethod
     def from_csv(cls, original_record: Dict[str, str]) -> "Entity":
-
         # clean the record
         record: Dict[str, Any] = {
             k: None if v.strip() in ["", "n/a"] else v.strip()

findthatpostcode/documents/postcode.py

@@ -79,23 +79,25 @@ def from_csv(cls, original_record: Dict[str, str]) -> "Postcode":
                 record[k] = None
 
         # date fields
-        for field in ["dointr", "doterm"]:
-            if record[field]:
-                record[field] = datetime.datetime.strptime(record[field], "%Y%m")
+        for date_field in ["dointr", "doterm"]:
+            if record[date_field]:
+                record[date_field] = datetime.datetime.strptime(
+                    record[date_field], "%Y%m"
+                )
 
         # latitude and longitude
-        for field in ["lat", "long"]:
-            if record[field]:
-                record[field] = float(record[field])
-                if record[field] == 99.999999:
-                    record[field] = None
+        for geo_field in ["lat", "long"]:
+            if record[geo_field]:
+                record[geo_field] = float(record[geo_field])
+                if record[geo_field] == 99.999999:
+                    record[geo_field] = None
         if record["lat"] and record["long"]:
             record["location"] = {"lat": record["lat"], "lon": record["long"]}
 
         # integer fields
-        for field in ["oseast1m", "osnrth1m", "usertype", "osgrdind", "imd"]:
-            if record[field]:
-                record[field] = int(record[field])
+        for int_field in ["oseast1m", "osnrth1m", "usertype", "osgrdind", "imd"]:
+            if record[int_field]:
+                record[int_field] = int(record[int_field])
 
         # add postcode hash
         record["hash"] = hashlib.md5(

findthatpostcode/documents/uprn.py

@@ -1,13 +1,6 @@
 from elasticsearch_dsl import (
-    Boolean,
-    Completion,
-    Date,
     Document,
-    InnerDoc,
     Keyword,
-    Nested,
-    Text,
-    analyzer,
 )
 
 

findthatpostcode/main.py

@@ -1,15 +1,14 @@
 import datetime
+from typing import Any
 
 from elasticsearch import Elasticsearch
 from fastapi import Depends, FastAPI, Request
 from fastapi.responses import HTMLResponse
 from fastapi.staticfiles import StaticFiles
 from fastapi.templating import Jinja2Templates
-from sqlalchemy.orm import Session
 
 from findthatpostcode import api, crud, graphql, settings
 from findthatpostcode.db import get_db
-from findthatpostcode.utils import is_latlon, is_postcode
 
 description = """
 This site presents data on UK postcodes and geographical areas, based on open data released by
@@ -56,7 +55,7 @@
             "name": "Get postcode",
             "description": "Postcode lookup",
         }
-    ]
+    ],
     # license_info={
     #     "name": "Apache 2.0",
     #     "url": "https://www.apache.org/licenses/LICENSE-2.0.html",
@@ -81,35 +80,37 @@
 
 @app.get("/", response_class=HTMLResponse, include_in_schema=False)
 def index(request: Request):
-    return templates.TemplateResponse("index.html.j2", {"request": request})
+    return templates.TemplateResponse(request, "index.html.j2")
 
 
 @app.get("/search/", response_class=HTMLResponse, include_in_schema=False)
 def search(request: Request, db: Elasticsearch = Depends(get_db)):
+    print(db)
     q = request.query_params.get("q", "")
-    context = {"request": request, "q": q}
+    context: dict[str, Any] = {"q": q}
     if q:
         areas = crud.search_areas(db, q)
         context["result"] = list(zip(areas.result, areas.scores))
         context["total"] = areas.result_count
-    return templates.TemplateResponse("areasearch.html.j2", context)
+    return templates.TemplateResponse(request, "areasearch.html.j2", context)
 
 
-@app.get("/areastypes/", response_class=HTMLResponse, include_in_schema=False)
+@app.get("/areatypes/", response_class=HTMLResponse, include_in_schema=False)
 def all_areatypes(request: Request, db: Elasticsearch = Depends(get_db)):
     return templates.TemplateResponse(
+        request,
         "areatypes.html.j2",
-        {"request": request, "areatypes": crud.get_all_areatypes(db)},
+        {"areatypes": crud.get_all_areatypes(db)},
     )
 
 
-@app.get("/areastypes/{areacode}", response_class=HTMLResponse, include_in_schema=False)
+@app.get("/areatypes/{areacode}", response_class=HTMLResponse, include_in_schema=False)
 def get_areatype(areacode: str, request: Request, db: Elasticsearch = Depends(get_db)):
     areatype = settings.AREA_TYPES.get(areacode)
     return templates.TemplateResponse(
+        request,
         "areatype.html.j2",
         {
-            "request": request,
             "result": {
                 **areatype,
                 "attributes": {
@@ -125,4 +126,4 @@ def get_areatype(areacode: str, request: Request, db: Elasticsearch = Depends(ge
 
 @app.get("/addtocsv/", response_class=HTMLResponse, include_in_schema=False)
 def addtocsv(request: Request):
-    return templates.TemplateResponse("addtocsv.html.j2", {"request": request})
+    return templates.TemplateResponse(request, "addtocsv.html.j2")