change everything to fastapi

Author: drkane

findthatpostcode/blueprints/__init__.py

@@ -1,3 +1,5 @@
+from fastapi import FastAPI
+
 from findthatpostcode.blueprints import (
     addtocsv,
     areas,
@@ -10,14 +12,14 @@
     tools,
 )
 
+app = FastAPI()
 
-def init_app(app):
-    app.register_blueprint(areas.bp)
-    app.register_blueprint(areatypes.bp)
-    app.register_blueprint(postcodes.bp)
-    app.register_blueprint(points.bp)
-    app.register_blueprint(reconcile.bp)
-    app.register_blueprint(addtocsv.bp)
-    app.register_blueprint(places.bp)
-    app.register_blueprint(search.bp)
-    app.register_blueprint(tools.bp)
+app.include_router(areas.bp)
+app.include_router(addtocsv.bp)
+app.include_router(areatypes.bp)
+app.include_router(places.bp)
+app.include_router(points.bp)
+# app.include_router(postcodes.bp)
+# app.include_router(reconcile.bp)
+# app.include_router(search.bp)
+# app.include_router(tools.bp)

findthatpostcode/blueprints/addtocsv.py

@@ -1,25 +1,28 @@
 import codecs
 import os
 import tempfile
+from typing import Annotated
 
-from flask import Blueprint, make_response, render_template, request
+from fastapi import APIRouter, Form, UploadFile
 
+from findthatpostcode.blueprints.areas import CSVResponse
 from findthatpostcode.blueprints.process_csv import process_csv
 from findthatpostcode.controllers.areas import area_types_count
-from findthatpostcode.db import get_db
+from findthatpostcode.db import ElasticsearchDep
 from findthatpostcode.metadata import (
     BASIC_UPLOAD_FIELDS,
     DEFAULT_UPLOAD_FIELDS,
     STATS_FIELDS,
 )
+from findthatpostcode.utils import templates
 
-bp = Blueprint("addtocsv", __name__, url_prefix="/addtocsv")
+bp = APIRouter(prefix="/addtocsv")
 
 
-@bp.route("/", strict_slashes=False, methods=["GET"])
-def addtocsv():
-    ats = area_types_count(get_db())
-    return render_template(
+@bp.get("/")
+def addtocsv(es: ElasticsearchDep):
+    ats = area_types_count(es)
+    return templates.TemplateResponse(
         "addtocsv.html.j2",
         result=ats,
         basic_fields=BASIC_UPLOAD_FIELDS,
@@ -28,11 +31,13 @@ def addtocsv():
     )
 
 
-@bp.route("/", strict_slashes=False, methods=["POST"])
-def return_csv():
-    upload = request.files.get("csvfile")
-    column_name = request.form.get("column_name", "postcode")
-    fields = request.form.getlist("fields")
+@bp.post("/")
+def return_csv(
+    es: ElasticsearchDep,
+    csvfile: UploadFile,
+    column_name: Annotated[str, Form()],
+    fields: Annotated[list[str], Form()],
+):
     if not fields:
         fields = DEFAULT_UPLOAD_FIELDS
 
@@ -56,23 +61,23 @@ def return_csv():
         fields.append("lep2_name")
         fields.remove("lep_name")
 
-    _, ext = os.path.splitext(upload.filename)
+    _, ext = os.path.splitext(csvfile.filename)
     if ext not in [".csv"]:
         return "File extension not allowed."
 
     with tempfile.SpooledTemporaryFile(mode="w+", newline="") as output:
         process_csv(
-            codecs.iterdecode(upload.stream, "utf-8"),
+            codecs.iterdecode(csvfile.file, "utf-8"),
             output,
-            get_db(),
+            es,
             column_name,
             fields,
         )
         output.seek(0)
 
-        response = make_response(output.read())
+        response = CSVResponse(output.read())
         response.headers["Content-Type"] = "text/csv"
         response.headers["Content-Disposition"] = 'attachment; filename="{}"'.format(
-            upload.filename
+            csvfile.filename
         )
         return response

findthatpostcode/blueprints/areas.py

@@ -2,37 +2,30 @@
 import io
 import re
 
-from flask import (
-    Blueprint,
-    abort,
-    jsonify,
-    make_response,
-    redirect,
-    render_template,
-    request,
-    url_for,
-)
+from fastapi import APIRouter, Request
+from fastapi.responses import JSONResponse, RedirectResponse
 
 from findthatpostcode.blueprints.utils import return_result
 from findthatpostcode.controllers.areas import Area, get_all_areas
-from findthatpostcode.db import get_db
+from findthatpostcode.db import ElasticsearchDep
+from findthatpostcode.utils import CSVResponse, templates
 
-bp = Blueprint("areas", __name__, url_prefix="/areas")
+bp = APIRouter(prefix="/areas")
 
 
-@bp.route("/search")
-@bp.route("/search.<filetype>")
-def area_search(filetype="json"):
-    return redirect(url_for("search.search_index", q=request.values.get("q")), code=301)
+@bp.get("/search")
+@bp.get("/search.<filetype>")
+def area_search(request: Request, filetype="json", q: str | None = None):
+    return RedirectResponse(
+        request.url_for("search.search_index", q=q), status_code=301
+    )
 
 
-@bp.route("/names.csv")
-def all_names():
+@bp.get("/names.csv")
+def all_names(es: ElasticsearchDep, types: str = ""):
     areas = get_all_areas(
-        get_db(),
-        areatypes=[
-            a.strip().lower() for a in request.values.get("types", "").split(",") if a
-        ],
+        es,
+        areatypes=[a.strip().lower() for a in types.split(",") if a],
     )
     return areas_csv(areas, "names.csv")
 
@@ -49,15 +42,14 @@ def areas_csv(areas, filename):
         if re.match(r"[A-Z][0-9]{8}", row.get("code")):
             writer.writerow(row)
 
-    output = make_response(si.getvalue())
+    output = CSVResponse(si.getvalue())
     output.headers["Content-Disposition"] = "attachment; filename={}".format(filename)
     output.headers["Content-type"] = "text/csv"
     return output
 
 
-@bp.route("/<areacodes>.geojson")
-def get_area_boundary(areacodes):
-    es = get_db()
+@bp.get("/<areacodes>.geojson")
+def get_area_boundary(areacodes: str, es: ElasticsearchDep):
     areacodes = areacodes.split("+")
     features = []
     for areacode in areacodes:
@@ -66,13 +58,12 @@ def get_area_boundary(areacodes):
         if status == 200:
             features.extend(r.get("features"))
     if status != 200:
-        return abort(make_response(jsonify(message=r), status))
-    return jsonify({"type": "FeatureCollection", "features": features})
+        return JSONResponse(message=r, status_code=status)
+    return JSONResponse({"type": "FeatureCollection", "features": features})
 
 
-@bp.route("/<areacode>/children/<areatype>.geojson")
-def get_area_children_boundary(areacode, areatype):
-    es = get_db()
+@bp.get("/<areacode>/children/<areatype>.geojson")
+def get_area_children_boundary(areacode: str, areatype: str, es: ElasticsearchDep):
     area = Area.get_from_es(areacode, es, boundary=False, examples_count=0)
     features = []
     errors = {}
@@ -84,31 +75,36 @@ def get_area_children_boundary(areacode, areatype):
         else:
             errors[child_area.id] = r
     if not features:
-        return abort(make_response(jsonify(message=errors), status))
-    return jsonify({"type": "FeatureCollection", "features": features})
+        return JSONResponse(message=errors, status_code=status)
+    return JSONResponse({"type": "FeatureCollection", "features": features})
 
 
 @bp.route("/<areacode>")
 @bp.route("/<areacode>.<filetype>")
-def get_area(areacode, filetype="json"):
+def get_area(
+    areacode: str,
+    filetype: str = "json",
+    es: ElasticsearchDep = None,
+    child: str | None = None,
+):
     result = Area.get_from_es(
         areacode,
-        get_db(),
+        es,
         boundary=(filetype == "geojson"),
         examples_count=(0 if filetype == "geojson" else 5),
     )
 
     if filetype == "geojson":
         status, r = result.geoJSON()
         if status != 200:
-            return abort(make_response(jsonify(message=r), status))
-        return jsonify(r)
+            return JSONResponse(message=r, status_code=status)
+        return JSONResponse(r)
 
     return return_result(
         result,
         filetype,
         "area.html.j2",
-        child=request.values.get("child"),
+        child=child,
         example_postcode_json=[
             p.attributes.get("location")
             for p in result.relationships["example_postcodes"]
@@ -119,4 +115,4 @@ def get_area(areacode, filetype="json"):
 
 @bp.route("/<areacodes>/map")
 def get_area_map(areacodes):
-    return render_template("area_map.html.j2", areacodes=areacodes)
+    return templates.TemplateResponse("area_map.html.j2", {"areacodes": areacodes})

findthatpostcode/blueprints/areatypes.py

@@ -1,23 +1,28 @@
-from flask import Blueprint, render_template, request, url_for
+from fastapi import APIRouter, Request
 
 from findthatpostcode.blueprints.areas import areas_csv
 from findthatpostcode.blueprints.utils import return_result
-from findthatpostcode.controllers.areas import Areatype, area_types_count, get_all_areas
+from findthatpostcode.controllers.areas import (
+    Areatype,
+    area_types_count,
+    get_all_areas,
+)
 from findthatpostcode.controllers.controller import Pagination
 from findthatpostcode.db import get_db
+from findthatpostcode.utils import templates
 
-bp = Blueprint("areatypes", __name__, url_prefix="/areatypes")
+bp = APIRouter(prefix="/areatypes")
 
 
-@bp.route("/", strict_slashes=False)
+@bp.get("/")
 def all():
     ats = area_types_count(get_db())
-    return render_template("areatypes.html.j2", result=ats)
+    return templates.TemplateResponse("areatypes.html.j2", {"result": ats})
 
 
-@bp.route("/<areacode>")
-@bp.route("/<areacode>.<filetype>")
-def get_areatype(areacode, filetype="json"):
+@bp.get("/<areacode>")
+@bp.get("/<areacode>.<filetype>")
+def get_areatype(areacode: str, request: Request, filetype: str = "json"):
     if filetype == "csv":
         areas = get_all_areas(get_db(), areatypes=[areacode.strip().lower()])
         return areas_csv(areas, "{}.csv".format(areacode))
@@ -27,7 +32,7 @@ def get_areatype(areacode, filetype="json"):
 
     pagination.set_pagination(result.attributes["count_areas"])
     nav = {
-        p: url_for(
+        p: request.url_for(
             "areatypes.get_areatype", areacode=areacode, filetype=filetype, **args
         )
         if isinstance(args, dict)

findthatpostcode/blueprints/places.py

@@ -1,25 +1,23 @@
-from flask import Blueprint, jsonify, redirect, request, url_for
+from fastapi import APIRouter, Request
+from fastapi.responses import JSONResponse, RedirectResponse
 
 from findthatpostcode.blueprints.utils import return_result
 from findthatpostcode.controllers.places import Place
-from findthatpostcode.db import get_db
+from findthatpostcode.db import ElasticsearchDep
 
-bp = Blueprint("places", __name__, url_prefix="/places")
+bp = APIRouter(prefix="/places")
 
 
-@bp.route("/redirect")
-def point_redirect():
-    lat = float(request.args.get("lat"))
-    lon = float(request.args.get("lon"))
-    return redirect(
-        url_for("places.nearest", lat=lat, lon=lon, filetype="html"), code=303
+@bp.get("/redirect")
+def point_redirect(lat: float, lon: float, request: Request):
+    return RedirectResponse(
+        request.url_for("places.nearest", lat=lat, lon=lon, filetype="html"), code=303
     )
 
 
-@bp.route("/nearest/<lat>,<lon>")
-@bp.route("/nearest/<lat>,<lon>.<filetype>")
-def nearest(lat, lon, filetype="json"):
-    es = get_db()
+@bp.get("/nearest/{lat},{lon}")
+@bp.get("/nearest/{lat},{lon}.{filetype}")
+def nearest(lat: float, lon: float, es: ElasticsearchDep, filetype: str = "json"):
     query = {
         "query": {"match_all": {}},
         "sort": [
@@ -34,11 +32,11 @@ def nearest(lat, lon, filetype="json"):
         size=10,
         _source_excludes=[],
     )
-    return jsonify(data)
+    return JSONResponse(content=data)
 
 
-@bp.route("/<areacode>")
-@bp.route("/<areacode>.<filetype>")
-def get_place(areacode, filetype="json"):
-    result = Place.get_from_es(areacode, get_db())
+@bp.get("/{areacode}")
+@bp.get("/{areacode}.{filetype}")
+def get_place(areacode: str, es: ElasticsearchDep, filetype: str = "json"):
+    result = Place.get_from_es(areacode, es)
     return return_result(result, filetype, "place.html.j2")

findthatpostcode/blueprints/points.py

@@ -1,31 +1,32 @@
-from flask import Blueprint, redirect, request, url_for
+from fastapi import APIRouter, Request
+from fastapi.responses import RedirectResponse
 
 from findthatpostcode.blueprints.utils import return_result
 from findthatpostcode.controllers.points import Point
-from findthatpostcode.db import get_db
+from findthatpostcode.db import ElasticsearchDep
 
-bp = Blueprint("points", __name__, url_prefix="/points")
+bp = APIRouter(prefix="/points")
 
 
 @bp.route("/redirect")
-def point_redirect():
-    lat = float(request.args.get("lat"))
-    lon = float(request.args.get("lon"))
-    return redirect(
-        url_for("points.get", latlon="{},{}.html".format(lat, lon)), code=303
+def point_redirect(lat: float, lon: float, request: Request):
+    return RedirectResponse(
+        request.url_for(
+            "points.get", latlon="{},{}.html".format(lat, lon), filetype="html"
+        ),
+        code=303,
     )
 
 
-@bp.route("/<latlon>")
-def get(latlon):
+@bp.get("/{latlon}")
+def get(latlon, es: ElasticsearchDep):
     filetype = "json"
     if latlon.endswith(".json"):
         latlon = latlon[:-5]
     elif latlon.endswith(".html"):
         latlon = latlon[:-5]
         filetype = "html"
     lat, lon = latlon.split(",")
-    es = get_db()
     result = Point.get_from_es((float(lat), float(lon)), es)
     errors = result.get_errors()
     if errors: