Appeasing Pylance.
This commit is contained in:
parent
a336442a92
commit
4e15972275
1 changed files with 36 additions and 19 deletions
|
|
@ -1,5 +1,6 @@
|
|||
from flask import Blueprint, request, render_template, jsonify, abort
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from typing import Any, Optional, List, cast, Type, Iterable
|
||||
|
||||
from .defaults import (
|
||||
default_query, default_create, default_update, default_delete, default_serialize
|
||||
|
|
@ -12,7 +13,7 @@ bp = Blueprint("ui", __name__, url_prefix="/ui")
|
|||
def _normalize(s: str) -> str:
|
||||
return s.replace("_", "").replace("-", "").lower()
|
||||
|
||||
def get_model_class(model_name: str):
|
||||
def get_model_class(model_name: str) -> type:
|
||||
"""Resolve a model class by name across SA/Flask-SA versions."""
|
||||
target = _normalize(model_name)
|
||||
|
||||
|
|
@ -36,7 +37,7 @@ def get_model_class(model_name: str):
|
|||
|
||||
abort(404, f"Unknown resource '{model_name}'")
|
||||
|
||||
def call(Model, name, *args, **kwargs):
|
||||
def call(Model: type, name: str, *args: Any, **kwargs: Any) -> Any:
|
||||
fn = getattr(Model, name, None)
|
||||
return fn(*args, **kwargs) if callable(fn) else None
|
||||
|
||||
|
|
@ -45,19 +46,28 @@ def list_items(model_name):
|
|||
Model = get_model_class(model_name)
|
||||
text = (request.args.get("q") or "").strip() or None
|
||||
limit_param = request.args.get("limit")
|
||||
limit = None if limit_param in (None, "", "0", "-1") else min(int(limit_param), 500)
|
||||
# limit = min(int(request.args.get("limit", 100)), 500)
|
||||
limit: int | None = None if limit_param in (None, "", "0", "-1") else min(int(limit_param), 500)
|
||||
offset = int(request.args.get("offset", 0))
|
||||
view = (request.args.get("view") or "json").strip()
|
||||
|
||||
rows = call(Model, "ui_query", db.session, text=text, limit=limit, offset=offset) \
|
||||
or default_query(db.session, Model, text=text, limit=limit, offset=offset)
|
||||
items = [call(Model, 'ui_serialize', r, view=view) or default_serialize(Model, r, view=view)
|
||||
for r in rows]
|
||||
# Build kwargs so we only include 'limit' when it's an int
|
||||
qkwargs: dict[str, Any] = {"text": text, "offset": offset}
|
||||
if limit is not None:
|
||||
qkwargs["limit"] = limit
|
||||
|
||||
rows_iter: Iterable[Any] = (
|
||||
call(Model, "ui_query", db.session, **qkwargs)
|
||||
or default_query(db.session, Model, **qkwargs)
|
||||
)
|
||||
rows = list(rows_iter)
|
||||
|
||||
items = [
|
||||
(call(Model, "ui_serialize", r, view=view) or default_serialize(Model, r, view=view))
|
||||
for r in rows
|
||||
]
|
||||
|
||||
want_option = (request.args.get("view") == "option")
|
||||
want_list = (request.args.get("view") == "list")
|
||||
print(view)
|
||||
if want_option:
|
||||
return render_template("fragments/_option_fragment.html", options=items)
|
||||
if want_list:
|
||||
|
|
@ -67,7 +77,7 @@ def list_items(model_name):
|
|||
@bp.post("/<model_name>/create")
|
||||
def create_item(model_name):
|
||||
Model = get_model_class(model_name)
|
||||
payload = request.get_json(silent=True) or {}
|
||||
payload: dict[str, Any] = request.get_json(silent=True) or {}
|
||||
if not payload:
|
||||
return jsonify({"error": "Payload required"}), 422
|
||||
try:
|
||||
|
|
@ -85,25 +95,32 @@ def create_item(model_name):
|
|||
@bp.post("/<model_name>/update")
|
||||
def update_item(model_name):
|
||||
Model = get_model_class(model_name)
|
||||
payload = request.get_json(silent=True) or {}
|
||||
try:
|
||||
id_ = int(payload.get("id"))
|
||||
except Exception:
|
||||
payload: dict[str, Any] = request.get_json(silent=True) or {}
|
||||
|
||||
id_raw: Any = payload.get("id")
|
||||
if isinstance(id_raw, bool): # bool is an int subclass; explicitly ban
|
||||
return jsonify({"error": "Invalid id"}), 422
|
||||
try:
|
||||
id_ = int(id_raw) # will raise on None, '', junk
|
||||
except (TypeError, ValueError):
|
||||
return jsonify({"error": "Invalid id"}), 422
|
||||
|
||||
obj = call(Model, 'ui_update', db.session, id_=id_, payload=payload) \
|
||||
or default_update(db.session, Model, id_, payload)
|
||||
if not obj:
|
||||
return jsonify({"error": "Note found"}), 404
|
||||
return jsonify({"error": "Not found"}), 404
|
||||
return ("", 204)
|
||||
|
||||
@bp.post("/<model_name>/delete")
|
||||
def delete_item(model_name):
|
||||
Model = get_model_class(model_name)
|
||||
payload = request.get_json(silent=True) or {}
|
||||
ids = payload.get("ids") or []
|
||||
payload: dict[str, Any] = request.get_json(silent=True) or {}
|
||||
ids_raw = payload.get("ids") or []
|
||||
if not isinstance(ids_raw, list):
|
||||
return jsonify({"error": "Invalid ids"}), 422
|
||||
try:
|
||||
ids = [int(x) for x in ids]
|
||||
except Exception:
|
||||
ids: List[int] = [int(x) for x in ids_raw]
|
||||
except (TypeError, ValueError):
|
||||
return jsonify({"error": "Invalid ids"}), 422
|
||||
try:
|
||||
deleted = call(Model, 'ui_delete', db.session, ids=ids) \
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue