ประสบการณ์การทำ API ด้วย Python

ได้มีโอกาสเขียน code ทำ API สำหรับการเข้าถึงฐานข้อมูล Postgresql ด้วย Python Flask เป็นประสบการณ์การเรียนรู้ที่ดี นอกจากเขียน code แล้ว ก็ได้มีโอกาสใช้ API Management tool อย่าง Kong

โจทย์

ทำ API ให้สามารถเพิ่มข้อมูล ลบข้อมูล เรียกดูข้อมูลจากตารางข้อมูลที่ชื่อว่า mobileinfo โดยมีโครงสร้างดังนี้

CREATE TABLE public.mobileinfo
(
    transactionid serial NOT NULL,
    userid integer NOT NULL,
    deviceid character varying(100) NOT NULL,
    latitude numeric(12, 9),
    longtitude numeric(12, 9),
    createdate date without time zone,
    mcc character varying(50),
    mnc character varying(50),
    cicid integer,
    taclac integer,
    rssi integer,
    mod integer,
    enodeb integer,
    eci integer,
    created_on timestamptz default now(),
    PRIMARY KEY (transactionid)
);

เรื่องราว

เนื่องจากเป็นคนที่ขยันไม่มาก จึงยอมเสียเวลาเสาะแสวงหา library ที่สามารถอ่านและเขียนข้อมูลจากรูปแบบของ JSON ลงในฐานข้อมูลได้เลย ก็พบบทความเก่าตั้งแต่ปี 2013 ว่า psycopg2 นั้นมี function/class ใน extras ที่สามารถทำได้ อ่านเอกสารอ้างอิง https://www.psycopg.org/docs/extras.html ก็ไม่เข้าใจ จึงต้องลงมือทดลองทำตามตัวอย่างในบทความ https://www.peterbe.com/plog/from-postgres-to-json-strings

ติดตั้ง psycopg2 ด้วย pip

pip install psycopg2-binary

code ที่นำมาเล่านี้ไม่ได้ครบถ้วนสมบูรณ์ทุกตัวอักษร จะเน้นเฉพาะส่วนของ psycopg2.extras.RealDictCursor เท่านั้น ตัวอย่าง code มีการใช้ Flask Blueprint, Jsonify, และอื่นๆ อีกมากมาย

CODE

เริ่มต้นก็ต้องทำการ import RealDictCursor กันก่อน

from psycopg2.extras import RealDictCursor

สำหรับการเรียกข้อมูลจากฐานข้อมูล ก็สร้าง route ขึ้นมา โดยในตัวอย่างนี้จะเรียกข้อมูลมาเพียง 10 รายการ ด้วยคำสั่ง SELECT * แล้วทำการส่งข้อมูลกลับในรูปแบบของ JSON

@svcmobile.route('/', methods=['GET'])
def svcmobile_main():
    sqlcmd = f"SELECT * FROM {DBTABLE} LIMIT 10;"
    logging.info(sqlcmd)
    db_conn = db_connect()
    with db_conn.cursor(cursor_factory=RealDictCursor) as curs:
        curs.execute(sqlcmd)
        results = curs.fetchall()
        if not results:
            msginfo = 'No records.'
            logging.info(msginfo)
            return jsonify({"status": msginfo})
    db_conn.close()
    return jsonify({"data":results}), 200

สำหรับการเพิ่มข้อมูลลงในฐานข้อมูล ก็จะเป็นอีก route ตามตัวอย่าง code ด้านล่างนี้

@svcmobile.route('/', methods=['POST'])
def svcmobile_add():
    payload = request.json
    msginfo = f'Payload: {payload}'
    logging.info(msginfo)

    # validate data
    if not payload:
        msginfo = 'No data provide!'
        logging.info(msginfo)
        return jsonify({"status": msginfo})
    if not "deviceid" in payload.keys():
        msginfo = 'No deviceid provided!'
        logging.info(msginfo)
        return jsonify({"status": msginfo})
    if not "userid" in payload.keys():
        msginfo = 'No userid provided!'
        logging.info(msginfo)
        return jsonify({"status": msginfo})

    db_conn = db_connect()
    # insert data
    sqlcmd = f"INSERT INTO {DBTABLE} (userid, deviceid, latitude, longtitude, createdate, mcc, mnc, cicid, taclac, rssi, mod, enodeb, eci) \
            VALUES ('{payload['userid']}', '{payload['deviceid']}', {payload['latitude']}, {payload['longtitude']}, '{payload['createdate']}', \
            '{payload['mcc']}', '{payload['mnc']}', {payload['cicid']}, {payload['taclac']}, {payload['rssi']}, {payload['mod']}, \
            {payload['enodeb']}, {payload['eci']});"
    logging.info(sqlcmd)
    with db_conn.cursor() as curs:
        try:
            curs.execute(sqlcmd)
            db_conn.commit()
        except:
            db_conn.rollback()
            return jsonify({'status':'Fail to save!'})

    db_conn.close()

    return jsonify({'status':'OK'}), 200