1. Tổng quan.
Viết API để cập nhật dữ liệu có trong cơ sở dữ liệu với Flask là một phần quan trọng của phát triển ứng dụng web.
Đây là một số lý thuyết và khái niệm quan trọng về cách thực hiện điều này:
- HTTP Methods: Trong API, bạn sẽ sử dụng các phương thức HTTP để chỉ định loại hoạt động bạn muốn thực hiện trên dữ liệu. Các phương thức chính cho cập nhật dữ liệu bao gồm:
- PUT: Sử dụng để cập nhật hoàn toàn một bản ghi có sẵn.
- PATCH: Sử dụng để cập nhật một phần của bản ghi có sẵn.
- POST: Sử dụng để tạo một bản ghi mới hoặc thực hiện một số thay đổi không rõ ràng.
- Endpoint: Mỗi API sẽ có các điểm cuối (endpoints) riêng, là các URL đại diện cho các tài nguyên cụ thể bạn muốn cập nhật.
- Validation: Trước khi cập nhật dữ liệu, bạn nên thực hiện kiểm tra dữ liệu đầu vào để đảm bảo tính hợp lệ và an toàn. Flask cung cấp các công cụ cho việc này.
- Database Interaction: Bạn sẽ sử dụng ORM (Object-Relational Mapping) hoặc SQLALCHEMY để tương tác với cơ sở dữ liệu. Đảm bảo rằng bạn đang sử dụng các truy vấn thích hợp để cập nhật dữ liệu.
- Authentication và Authorization: Bảo mật là một yếu tố quan trọng. Đảm bảo rằng chỉ người dùng được ủy quyền mới có thể thực hiện các thay đổi trong dữ liệu.
- Error Handling: Xử lý lỗi là một phần quan trọng. Đảm bảo rằng bạn xử lý tất cả các tình huống lỗi một cách thích hợp và trả về mã lỗi thích hợp.
Dưới đây là một ví dụ đơn giản về cách viết một API để cập nhật 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 Item(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(100), nullable=False)
description = db.Column(db.String(255))
@app.route('/items/<int:item_id>', methods=['PUT'])
def update_item(item_id):
item = Item.query.get(item_id)
if item is None:
return jsonify({'message': 'Item not found'}), 404
data = request.get_json()
item.name = data['name']
item.description = data['description']
db.session.commit()
return jsonify({'message': 'Item updated successfully'})
if __name__ == '__main__':
app.run(debug=True)Trong ví dụ này, chúng ta đã xây dựng một API cho phép cập nhật thông tin của một mục cụ thể trong cơ sở dữ liệu. Chúng ta sử dụng phương thức PUT để thực hiện cập nhật. Dữ liệu đầu vào được truyền dưới dạng JSON, và chúng ta sử dụng SQLAlchemy để tương tác với cơ sở dữ liệu.
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, update_or_insert_arp_service, get_all_arp_from_device_service,
get_all_arp_from_db_service, get_arp_by_id_from_db_service, update_arp_entry_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- 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!!!"- 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_data3. Kiểm tra kết quả.
- Với API
http://192.168.13.200:5000/update_arp_entry.
Kiếm tra lại DB.
Nếu bạn giữ nguyên data và update lại, kết quả sẽ không có data nào được update vì không có sự thay đổi.
