103 lines
3.7 KiB
Python
103 lines
3.7 KiB
Python
"""
|
|
Entry point for running the Flask application.
|
|
"""
|
|
from sqlalchemy import not_, or_
|
|
from sqlalchemy.orm import Session
|
|
from app import app, db
|
|
from app.models.models import *
|
|
from app.models.network import *
|
|
|
|
def setup_database():
|
|
"""
|
|
Set up the database by creating all tables and initializing with default data.
|
|
"""
|
|
print("Running setup database")
|
|
with app.app_context():
|
|
db.create_all()
|
|
|
|
def get_network_ports_not_on_host(session: Session, network_id: str, exclude_host_id: str):
|
|
"""
|
|
Query all network ports on a network NOT assigned to a specific host.
|
|
|
|
Args:
|
|
session: SQLAlchemy session
|
|
network_id (str): Target network ID
|
|
exclude_host_id (str): Workload host ID to exclude
|
|
|
|
Returns:
|
|
List[NetworkPort]: Matching network ports
|
|
"""
|
|
return (
|
|
session.query(NetworkPort)
|
|
.filter(
|
|
NetworkPort.network_id == network_id,
|
|
not_(
|
|
session.query(Workload)
|
|
.filter(
|
|
Workload.id == NetworkPort.workload_id,
|
|
Workload.workload_host_id == exclude_host_id,
|
|
)
|
|
.exists()
|
|
),
|
|
not_(
|
|
NetworkPort.workload_id == None,
|
|
),
|
|
)
|
|
.all()
|
|
)
|
|
|
|
def print_port_details(session: Session, ports):
|
|
"""
|
|
Print details for each port including port_id, network_id, and workload_host_id.
|
|
"""
|
|
print("\nPort Details:")
|
|
print("{:<40} {:<40} {:<40} {:<40}".format("PORT ID", "NETWORK ID", "WORKLOAD_ID", "WORKLOAD HOST ID"))
|
|
print("-" * 120)
|
|
|
|
for port in ports:
|
|
workload_host_id = None
|
|
if port.workload_id:
|
|
# Modern SQLAlchemy 2.0 way to get a single object
|
|
workload = session.get(Workload, port.workload_id)
|
|
workload_host_id = workload.workload_host_id if workload else None
|
|
|
|
print("{:<40} {:<40} {:<40} {:<40}".format(
|
|
port.id,
|
|
port.network_id,
|
|
port.workload_id if port.workload_id else "None",
|
|
str(workload_host_id) if workload_host_id else "None"
|
|
))
|
|
|
|
if __name__ == '__main__':
|
|
with app.app_context():
|
|
# Create a new session
|
|
session = Session(db.engine)
|
|
|
|
try:
|
|
# Example IDs - replace with your actual values
|
|
network_id = "0a53f098-4583-449b-b041-140b32e52de1"
|
|
exclude_host_id = "85f2f58a-b217-4af4-8c6d-2348252f9009s"
|
|
|
|
# Get ports not on the excluded host
|
|
filtered_ports = get_network_ports_not_on_host(session, network_id, exclude_host_id)
|
|
print(f"\nFound {len(filtered_ports)} ports NOT on host {exclude_host_id}:")
|
|
print_port_details(session, filtered_ports)
|
|
|
|
# # Get all ports on the network
|
|
# all_ports = session.query(NetworkPort).filter(NetworkPort.network_id == network_id).all()
|
|
# print(f"\nAll {len(all_ports)} ports on network {network_id}:")
|
|
# print_port_details(session, all_ports)
|
|
|
|
# # Calculate and print statistics
|
|
# assigned_ports = [p for p in all_ports if p.workload_id]
|
|
# unassigned_ports = [p for p in all_ports if not p.workload_id]
|
|
|
|
# print("\nStatistics:")
|
|
# print(f"Total ports: {len(all_ports)}")
|
|
# print(f"Assigned to workloads: {len(assigned_ports)}")
|
|
# print(f"Unassigned: {len(unassigned_ports)}")
|
|
# print(f"Excluded from host {exclude_host_id}: {len(filtered_ports)}")
|
|
|
|
finally:
|
|
# Always close the session
|
|
session.close() |