1. Tổng quan.
Viết API kết hợp nhiều bảng trong cơ sở dữ liệu trong Flask là một tác vụ phức tạp nhưng quan trọng khi bạn muốn truy vấn và thao tác với dữ liệu từ nhiều bảng liên quan trong cơ sở dữ liệu của bạn.
Dưới đây là lý thuyết về cách viết một API này:
- ORM (Object-Relational Mapping): Flask thường sử dụng SQLAlchemy làm ORM để làm việc với cơ sở dữ liệu. Bạn cần định nghĩa các model (lớp) Python tương ứng với các bảng trong cơ sở dữ liệu. Mỗi model sẽ chứa đối tượng tương ứng với một bảng.
- Relationships (Mối quan hệ): Nếu các bảng trong cơ sở dữ liệu có mối quan hệ với nhau (ví dụ: quan hệ một-nhiều hoặc nhiều-nhiều), bạn cần xác định các mối quan hệ này trong định nghĩa model của bạn. SQLAlchemy hỗ trợ các mối quan hệ như ForeignKey và relationship.
- Routing: Bạn cần định tuyến các API endpoints để xử lý các yêu cầu từ client. Các endpoints này sẽ xác định cách bạn truy vấn và thao tác với dữ liệu từ nhiều bảng.
- Truy vấn và Thao tác: Khi bạn nhận được một yêu cầu từ client, bạn cần viết mã để thực hiện truy vấn và thao tác với cơ sở dữ liệu. Bạn có thể sử dụng SQLAlchemy để tạo các truy vấn phức tạp và thực hiện các thao tác như lấy dữ liệu từ nhiều bảng, cập nhật, xóa hoặc thêm mới.
- Trả về kết quả: Cuối cùng, bạn nên trả về kết quả của yêu cầu trong định dạng JSON hoặc XML để client có thể hiển thị hoặc sử dụng dữ liệu.
Dưới đây là một ví dụ về việc viết API kết hợp nhiều bảng trong Flask sử dụng SQLAlchemy:
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import joinedload
from sqlalchemy.exc import IntegrityError
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://username:password@localhost/mydatabase'
db = SQLAlchemy(app)
class Author(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(100))
books = db.relationship('Book', back_populates='author')
class Book(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(100))
author_id = db.Column(db.Integer, db.ForeignKey('author.id'))
author = db.relationship('Author', back_populates='books')
@app.route('/api/authors/<int:author_id>/books', methods=['GET'])
def get_books_by_author(author_id):
try:
author = Author.query.get(author_id)
if not author:
return jsonify({"message": "Author not found"}), 404
books = Book.query.filter_by(author_id=author_id).all()
book_data = [{"id": book.id, "title": book.title} for book in books]
return jsonify({"author": author.name, "books": book_data}), 200
except Exception as e:
return jsonify({"message": str(e)}), 500
if __name__ == '__main__':
app.run(debug=True)Trong ví dụ này, chúng ta có hai bảng: Author (Tác giả) và Book (Sách). Có một mối quan hệ một-nhiều giữa tác giả và sách, và chúng ta viết một API để lấy tất cả sách của một tác giả cụ thể.
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 và do mình không tạo nhiều bảng DB nên mình sẽ sử dụng phep so sánh Data giữa thiết bị và trong DB 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, compare_arp_data_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()
@get_arp_bl.route("/compare_arp_data", methods=['GET'])
def compare_arp_data():
return compare_arp_data_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)}"
def compare_arp_data_service():
device_arp_data = get_all_arp_from_device_service()
db_arp_data = get_all_arp_from_db_service()
result = {
"data_exists_in_both_but_has_different_values": [],
"data_exists_in_device_only": [],
"data_exists_in_db_only": [],
}
processed_addresses = set() # Danh sách các address đã được xử lý
# Tìm các dict có address tồn tại trong cả hai danh sách
for device_entry in device_arp_data:
address = device_entry['address']
if address not in processed_addresses:
db_entry = next((entry for entry in db_arp_data if entry['address'] == address), None)
if db_entry:
# So sánh các trường khác nhau (không bao gồm 'id')
different_fields = {}
for key in device_entry:
if key != 'id' and device_entry[key] != db_entry[key]:
different_fields[key] = device_entry[key]
if different_fields:
# Thêm cả device_entry và db_entry vào dictionary kết quả
entry_with_differences = {
"device_entry": device_entry,
"db_entry": db_entry,
}
result["data_exists_in_both_but_has_different_values"].append(entry_with_differences)
else:
# Nếu address chỉ tồn tại trong device_arp_data
result["data_exists_in_device_only"].append(device_entry)
# Đánh dấu address đã được xử lý
processed_addresses.add(address)
# Tìm các dict có address chỉ tồn tại trong db_arp_data
for db_entry in db_arp_data:
address = db_entry['address']
if address not in processed_addresses:
# Nếu address chỉ tồn tại trong db_arp_data
result["data_exists_in_db_only"].append(db_entry)
return result- 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ả.
Sử dụng API vừa tạo http://192.168.13.200:5000/compare_arp_data bạn sẽ nhận kết quả không có dữ liệu nào thay đổi giữa thiết bị và DB.
Giờ mình thực hiện xoá 1 giá trị trong DB.
Bạn hãy kiểm tra lại http://192.168.13.200:5000/compare_arp_data bạn sẽ thấy có 1 data tồn tại trên thiết bị nhưng trên DB không có nó.
Tiếp theo mình thực hiện thay đổi vlan76 sang vlan1000 cho ip 103.110.128.4.
Và kết quả khi sử dụng API http://192.168.13.200:5000/compare_arp_data là.
Bây giờ mình sẽ bỏ 2 giá trị trong thiết bị đó là.
{
"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"
}Và đây là kết quả.
