90 lines
2.8 KiB
Python
90 lines
2.8 KiB
Python
from flask import Blueprint, jsonify, request
|
|
|
|
from crudkit.api._cursor import encode_cursor, decode_cursor
|
|
from crudkit.core.service import _is_truthy
|
|
|
|
def generate_crud_blueprint(model, service):
|
|
bp = Blueprint(model.__name__.lower(), __name__)
|
|
|
|
@bp.get('/')
|
|
def list_items():
|
|
args = request.args.to_dict(flat=True)
|
|
|
|
# legacy detection
|
|
legacy_offset = "offset" in args or "page" in args
|
|
|
|
# sane limit default
|
|
try:
|
|
limit = int(args.get("limit", 50))
|
|
except Exception:
|
|
limit = 50
|
|
args["limit"] = limit
|
|
|
|
if legacy_offset:
|
|
# Old behavior: honor limit/offset, same CRUDSpec goodies
|
|
items = service.list(args)
|
|
return jsonify([obj.as_dict() for obj in items])
|
|
|
|
# New behavior: keyset seek with cursors
|
|
key, backward = decode_cursor(args.get("cursor"))
|
|
|
|
window = service.seek_window(
|
|
args,
|
|
key=key,
|
|
backward=backward,
|
|
include_total=_is_truthy(args.get("include_total", "1")),
|
|
)
|
|
|
|
desc_flags = list(window.order.desc)
|
|
body = {
|
|
"items": [obj.as_dict() for obj in window.items],
|
|
"limit": window.limit,
|
|
"next_cursor": encode_cursor(window.last_key, desc_flags, backward=False),
|
|
"prev_cursor": encode_cursor(window.first_key, desc_flags, backward=True),
|
|
"total": window.total,
|
|
}
|
|
|
|
resp = jsonify(body)
|
|
# Optional Link header
|
|
links = []
|
|
if body["next_cursor"]:
|
|
links.append(f'<{request.base_url}?cursor={body["next_cursor"]}&limit={window.limit}>; rel="next"')
|
|
if body["prev_cursor"]:
|
|
links.append(f'<{request.base_url}?cursor={body["prev_cursor"]}&limit={window.limit}>; rel="prev"')
|
|
if links:
|
|
resp.headers["Link"] = ", ".join(links)
|
|
return resp
|
|
|
|
@bp.get('/<int:id>')
|
|
def get_item(id):
|
|
item = service.get(id, request.args)
|
|
try:
|
|
return jsonify(item.as_dict())
|
|
except Exception as e:
|
|
return jsonify({"status": "error", "error": str(e)})
|
|
|
|
@bp.post('/')
|
|
def create_item():
|
|
obj = service.create(request.json)
|
|
try:
|
|
return jsonify(obj.as_dict())
|
|
except Exception as e:
|
|
return jsonify({"status": "error", "error": str(e)})
|
|
|
|
@bp.patch('/<int:id>')
|
|
def update_item(id):
|
|
obj = service.update(id, request.json)
|
|
try:
|
|
return jsonify(obj.as_dict())
|
|
except Exception as e:
|
|
return jsonify({"status": "error", "error": str(e)})
|
|
|
|
@bp.delete('/<int:id>')
|
|
def delete_item(id):
|
|
service.delete(id)
|
|
try:
|
|
return jsonify({"status": "success"}), 204
|
|
except Exception as e:
|
|
return jsonify({"status": "error", "error": str(e)})
|
|
|
|
return bp
|