Flask API with PUT/DELETE and Pagination, Search

Flask API with PUT/DELETE and Pagination, Search

In this episode, you’ll extend the Flask REST API to support full CRUD operations on your customers table. You will also add support for pagination and search filtering.

---

1. Dependencies (Same as Episode 3)

pip install flask ibm-db flask-cors

flask-cors helps avoid CORS errors when calling API from a frontend.

2. Updated app.py

from flask import Flask, request, jsonify
from flask_cors import CORS
import ibm_db

app = Flask(__name__)
CORS(app)

# Db2 connection string
conn_str = (
    "DATABASE=mydb;"
    "HOSTNAME=localhost;"
    "PORT=50000;"
    "PROTOCOL=TCPIP;"
    "UID=db2inst1;"
    "PWD=Passw0rd;"
)

try:
    conn = ibm_db.connect(conn_str, "", "")
except:
    raise Exception("Could not connect to Db2")

# --- GET with pagination and search ---
@app.route('/customers', methods=['GET'])
def get_customers():
    name = request.args.get('name')
    limit = int(request.args.get('limit', 10))
    offset = int(request.args.get('offset', 0))

    if name:
        sql = "SELECT id, name, email FROM customers WHERE name LIKE ? ORDER BY id LIMIT ? OFFSET ?"
        stmt = ibm_db.prepare(conn, sql)
        ibm_db.execute(stmt, (f"%{name}%", limit, offset))
    else:
        sql = "SELECT id, name, email FROM customers ORDER BY id LIMIT ? OFFSET ?"
        stmt = ibm_db.prepare(conn, sql)
        ibm_db.execute(stmt, (limit, offset))

    customers = []
    row = ibm_db.fetch_assoc(stmt)
    while row:
        customers.append(row)
        row = ibm_db.fetch_assoc(stmt)
    return jsonify(customers)

# --- POST: create new customer ---
@app.route('/customers', methods=['POST'])
def add_customer():
    data = request.get_json()
    sql = "INSERT INTO customers (id, name, email) VALUES (?, ?, ?)"
    stmt = ibm_db.prepare(conn, sql)
    try:
        ibm_db.execute(stmt, (data['id'], data['name'], data['email']))
        return jsonify({"message": "Customer added"}), 201
    except:
        return jsonify({"error": "Insert failed"}), 500

# --- PUT: update customer ---
@app.route('/customers/<int:customer_id>', methods=['PUT'])
def update_customer(customer_id):
    data = request.get_json()
    sql = "UPDATE customers SET name = ?, email = ? WHERE id = ?"
    stmt = ibm_db.prepare(conn, sql)
    try:
        ibm_db.execute(stmt, (data['name'], data['email'], customer_id))
        return jsonify({"message": "Customer updated"})
    except:
        return jsonify({"error": "Update failed"}), 500

# --- DELETE: remove customer ---
@app.route('/customers/<int:customer_id>', methods=['DELETE'])
def delete_customer(customer_id):
    sql = "DELETE FROM customers WHERE id = ?"
    stmt = ibm_db.prepare(conn, sql)
    try:
        ibm_db.execute(stmt, (customer_id,))
        return jsonify({"message": "Customer deleted"})
    except:
        return jsonify({"error": "Delete failed"}), 500

if __name__ == '__main__':
    app.run(debug=True)

3. Sample Requests

1. Search + Pagination

   GET /customers?name=Ali&limit=5&offset=10
   

2. Update Customer ```http PUT /customers/3 Content-Type: application/json

{ “name”: “Alice Updated”, “email”: “alice.new@example.com” }

3. Delete Customer
```http
DELETE /customers/3

4. Notes & Tips

Feature	Tip
Pagination	Always ORDER BY to ensure consistent results
LIKE search	Consider adding LOWER(name) for case-insensitive matching
DELETE	Make sure id is the primary key
UPDATE	Always validate input data before executing queries

Summary

Action	Method	Endpoint
Get all customers	GET	/customers
Search + paginate	GET	/customers?name=...&limit=...&offset=...
Add new customer	POST	/customers
Update customer	PUT	/customers/<id>
Delete customer	DELETE	/customers/<id>