[REST API] – FLASK PYTHON #6: Viết API Delete dữ liệu có trong Database

1. Tổng quan.

Viết một API để xóa dữ liệu từ cơ sở dữ liệu trong Flask là một phần quan trọng của việc quản lý ứng dụng web. Dưới đây là lý thuyết về cách viết một API Delete dữ liệu có trong cơ sở dữ liệu Flask:

HTTP Method: API Delete thường sử dụng phương thức HTTP DELETE. Điều này chỉ ra rằng bạn muốn xóa một mục cụ thể.
Định danh tài nguyên: Để xác định dữ liệu cần xóa, bạn cần cung cấp một cách để xác định tài nguyên cụ thể. Thông thường, bạn sẽ sử dụng một loại định danh như ID hoặc một trường duy nhất trong dữ liệu.
Routing: Bạn cần định tuyến API để lắng nghe yêu cầu DELETE. Điều này thường được thực hiện bằng cách sử dụng Blueprint trong Flask để tạo các tuyến đường.
Thao tác xóa dữ liệu: Khi API nhận yêu cầu DELETE, nó sẽ tìm tài nguyên cần xóa trong cơ sở dữ liệu, dựa vào định danh bạn cung cấp. Sau đó, nó sẽ xóa tài nguyên này khỏi cơ sở dữ liệu.
Trả về phản hồi: Sau khi xóa dữ liệu thành công, API nên trả về một thông báo xác nhận hoặc mã trạng thái HTTP thích hợp (thông thường là 204 No Content).

Dưới đây là một ví dụ về cách viết một API Delete dữ liệu từ cơ sở dữ liệu trong Flask:

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://username:password@localhost/mydatabase'
db = SQLAlchemy(app)

class MyData(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100))

@app.route('/api/data/<int:data_id>', methods=['DELETE'])
def delete_data(data_id):
    data = MyData.query.get(data_id)
    
    if data:
        db.session.delete(data)
        db.session.commit()
        return jsonify({"message": "Data deleted successfully"}), 204
    else:
        return jsonify({"message": "Data not found"}), 404

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

Trong ví dụ này, chúng ta định tuyến một API DELETE để xóa dữ liệu dựa trên data_id. Nếu dữ liệu tồn tại và được xóa thành công, chúng ta trả về mã trạng thái 204 (No Content). Nếu không tìm thấy dữ liệu, chúng ta trả về mã trạng thái 404 (Not Found).

2. Thực hành.

Ở phần này chúng ta chỉ làm các file theo cây thư mục này nhé.

devnet
├── cisco_os
│   ├── __init__.py
│   ├── backend
│   │   └── arp
│   │       ├── controller.py
│   │       ├── getdata.py
│   │       └── services.py
│   ├── config.py
│   ├── extension.py
│   └── model.py
├── network.py
└── venv
    ├── db_install.sh
    └── requirements.txt

config.py

import os
from dotenv import load_dotenv

load_dotenv()
SECRET_KEY = os.environ.get("KEY")
SQLALCHEMY_DATABASE_URI = os.environ.get("DATABASE_URL")
SQLALCHEMY_TRACK_MODIFICATIONS = False

.env

KEY = "hoanghd-secret-key"
DATABASE_URL = 'mysql://hoanghd:Hoanghd164@192.168.13.200/cisco_info'

__init__.py

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from .backend.arp.controller import get_arp_bl
from .model import db

def create_db(app):
    with app.app_context():
        db.create_all()
        print("Created DB!")

def create_cisco_os_app(config_file="config.py"):
    cisco_os = Flask(__name__)
    cisco_os.config.from_pyfile(config_file)
    db.init_app(cisco_os)
    create_db(cisco_os)
    cisco_os.register_blueprint(get_arp_bl)

    return cisco_os

controller.py

from flask import Blueprint, request
import json
from .services import (add_arp_service, get_all_arp_from_device_service, update_or_insert_arp_service, get_all_arp_from_db_service, get_arp_by_id_from_db_service,
                       update_arp_entry_service, delete_arp_service, delete_all_entries_service)

get_arp_bl = Blueprint("get_arp_bl", __name__)

@get_arp_bl.route("/import_all_arp", methods=['GET'])
def import_all_arp():
    return update_or_insert_arp_service()

@get_arp_bl.route("/get_all_arp_from_device", methods=['GET'])
def get_all_arp_from_device():
    return get_all_arp_from_device_service()

@get_arp_bl.route("/get_all_arp_from_db", methods=['GET'])
def get_all_arp_from_db():
    return get_all_arp_from_db_service()

@get_arp_bl.route('/get_arp/<int:entry_id>', methods=["GET"])
def route_get_arp_by_id(entry_id):
    arp_data = get_arp_by_id_from_db_service(entry_id)
    if arp_data:
        return json.dumps(arp_data), 200, {'Content-Type': 'application/json'}
    else:
        return "No ARP data found with the specified ID!!!", 404

@get_arp_bl.route("/add_arp", methods=['POST'])
def add_arp():
    return add_arp_service()

@get_arp_bl.route("/update_arp_entry", methods=['PUT'])
def update_arp_entry_route():
    data = request.get_json()
    if data:
        result = update_arp_entry_service(data)
        return result
    else:
        return "Invalid data!!!", 400

@get_arp_bl.route("/delete_arp", methods=['POST'])
def delete_arp():
    return delete_arp_service()

@get_arp_bl.route("/delete_all_arp", methods=['DELETE'])
def delete_all_entries():
    return delete_all_entries_service()

services.py

from flask import request
from .getdata import merged_arp_data
from ...model import db, cisco_arp

def get_all_arp_from_device_service():
    arp_data = merged_arp_data()
    return arp_data

def get_all_arp_from_db_service():
    all_arp_entries = cisco_arp.query.all()
    arp_data = []
    for entry in all_arp_entries:
        arp_data.append({
            "id": entry.id,
            "address": entry.address,
            "device": entry.device,
            "interface": entry.interface,
            "mac_address": entry.mac_address,
            "ports": entry.ports
        })
    return arp_data

def get_arp_by_id_from_db_service(entry_id):
    arp_entry = cisco_arp.query.filter_by(id=entry_id).first()
    if arp_entry:
        arp_data = {
            "id": arp_entry.id,
            "address": arp_entry.address,
            "device": arp_entry.device,
            "interface": arp_entry.interface,
            "mac_address": arp_entry.mac_address,
            "ports": arp_entry.ports
        }
        return arp_data
    else:
        return None

def update_or_insert_arp_service():
    arp_data = merged_arp_data()
    for item in arp_data:
        address = item['address']
        device = item['device']
        interface = item['interface']
        mac_address = item['mac_address']
        ports = item['ports']
        existing_entry = cisco_arp.query.filter_by(address=address).first()

        if existing_entry:
            existing_entry.device = device
            existing_entry.interface = interface
            existing_entry.mac_address = mac_address
            existing_entry.ports = ports
            print("Update %s into database sussess!!!" %(address))
        else:
            new_entry = cisco_arp(address=address, device=device, interface=interface, mac_address=mac_address, ports=ports)
            db.session.add(new_entry)
            print("Add %s into database sussess!!!" %(address))
        db.session.commit()
        
    return "Add all arp into database sussess!!!"

def add_arp_service():
    try:
        address = request.json['address']
        device = request.json['device']
        interface = request.json['interface']
        mac_address = request.json['mac_address']
        ports = request.json['ports']

        existing_arp = cisco_arp.query.filter_by(address=address).first()

        if existing_arp:
            return "ARP address already exists in the database."
        else:
            new_arp = cisco_arp(address=address, device=device, interface=interface, mac_address=mac_address, ports=ports)
            db.session.add(new_arp)
            db.session.commit()

            print("Successfully added %s to the database!!!" % (address))
            return "Added successfully!!!"
    except Exception as e:
        print("Error when adding ARP:", str(e))
        return "Error when adding ARP:", str(e)

def update_arp_entry_service(data):
    address = data["address"]
    arp_entry = cisco_arp.query.filter_by(address=address).first()
    
    if arp_entry:
        if (
            arp_entry.device != data["device"]
            or arp_entry.interface != data["interface"]
            or arp_entry.mac_address != data["mac_address"]
            or arp_entry.ports != data["ports"]
        ):
            arp_entry.device = data["device"]
            arp_entry.interface = data["interface"]
            arp_entry.mac_address = data["mac_address"]
            arp_entry.ports = data["ports"]
            db.session.commit()
            return "Data updated successfully!!!"
        else:
            return "No changes in the data!!!"
    else:
        return "No data found with the given address!!!"
    
def delete_arp_service():
    addresses_to_delete = request.get_json()
    
    if addresses_to_delete:
        for address in addresses_to_delete:
            entry_to_delete = cisco_arp.query.filter_by(address=address).first()
            if entry_to_delete:
                db.session.delete(entry_to_delete)
                db.session.commit()
                print("Delete %s into database sussess!!!" %(address))
            else:
                print("Address %s not found!!!" %(address))

    return "Delete Sussess!!!"

def delete_all_entries_service():
    try:
        cisco_arp.query.delete()
        db.session.commit()
        return "Delete all data sussess!!!"
    except Exception as e:
        db.session.rollback()
        return f"Error when delete data: {str(e)}"

model.py

from .extension import db

class cisco_arp(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    address = db.Column(db.String(255))
    device = db.Column(db.String(255))
    interface = db.Column(db.String(255))
    mac_address = db.Column(db.String(255))
    ports = db.Column(db.String(255))

    def __init__(self, address, device, interface, mac_address, ports):
        self.address = address
        self.device = device
        self.interface = interface
        self.mac_address = mac_address
        self.ports = ports

extension.py

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

network.py

from cisco_os import create_cisco_os_app

if __name__ == "__main__":
    cisco_network = create_cisco_os_app()
    cisco_network.run(debug=True, host='0.0.0.0', port=5000)

getdata.py

def merged_arp_data():
    arp_data = [
        {
            "address": "103.110.128.4",
            "device": "42.96.37.41",
            "interface": "Vlan76",
            "mac_address": "7cc2.5527.a594",
            "ports": "Po4"
        },
        {
            "address": "103.110.128.3",
            "device": "42.96.37.41",
            "interface": "Vlan76",
            "mac_address": "7cc2.5527.a8b0",
            "ports": "Po3"
        },
        {
            "address": "103.110.128.6",
            "device": "42.96.37.41",
            "interface": "Vlan76",
            "mac_address": "7cc2.5527.a974",
            "ports": "Po6"
        },
        {
            "address": "103.110.128.5",
            "device": "42.96.37.41",
            "interface": "Vlan76",
            "mac_address": "7cc2.5527.a9b4",
            "ports": "Po5"
        },
        {
            "address": "103.110.128.7",
            "device": "42.96.37.41",
            "interface": "Vlan76",
            "mac_address": "7cc2.5527.a9cc",
            "ports": "Po7"
        },
        {
            "address": "103.110.128.1",
            "device": "42.96.37.41",
            "interface": "Vlan76",
            "mac_address": "7cc2.5527.a9e8",
            "ports": "Po1"
        },
        {
            "address": "172.16.30.1",
            "device": "42.96.37.41",
            "interface": "Vlan1010",
            "mac_address": "0059.dc66.c1f5",
            "ports": "Vl1010"
        },
        {
            "address": "42.96.37.41",
            "device": "42.96.37.41",
            "interface": "Vlan214",
            "mac_address": "0059.dc66.c1fa",
            "ports": "Vl214"
        },
        {
            "address": "42.96.37.40",
            "device": "42.96.37.41",
            "interface": "Vlan214",
            "mac_address": "84eb.ef5d.461a",
            "ports": "Po100"
        },
        {
            "address": "42.96.37.43",
            "device": "42.96.37.41",
            "interface": "Vlan215",
            "mac_address": "0059.dc66.c1c8",
            "ports": "Vl215"
        },
        {
            "address": "42.96.37.42",
            "device": "42.96.37.41",
            "interface": "Vlan215",
            "mac_address": "f87a.4130.7a6a",
            "ports": "Po100"
        },
        {
            "address": "103.110.128.254",
            "device": "42.96.37.41",
            "interface": "Vlan76",
            "mac_address": "0059.dc66.c1e3",
            "ports": "Vl76"
        },
        {
            "address": "103.110.128.4",
            "device": "42.96.37.41",
            "interface": "Vlan76",
            "mac_address": "7cc2.5527.a594",
            "ports": "Po4"
        }]
    return arp_data

3. Kiểm tra kết quả.