Flask API with PUT/DELETE and Pagination, Search

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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)