ab301df9ac
Every network device on this fleet already follows one convention, applied by hand: AP-<name>, SW-<name>, SVR-<name>, IDF-<name>. 45 records, no exceptions. The create form demanded the asset number anyway, so the same value was typed twice and the convention held only as long as everyone remembered it. The prefix now lives on the device type, and a blank asset number is generated as <PREFIX>-<name>. Left explicit, an asset number always wins: a device carrying a real identifier of its own - a vendor tag, a controller name, a serial - keeps it. That is the platform rule, adopt where an identifier exists and derive only where none does. The prefix is NOT derived from the type name. "Access Point" and "Access Panel" both initialise to AP, and assetnumber is unique, so the second type would collide with the first on every device it created. It is nullable, so a type that wants no prefix generates the bare name rather than needing one invented. Names are sanitised before they reach a business key - the existing data already shows why, with IDF-Telco-Demarc-#1 carrying a '#' into an identifier. An existing prefix is never stacked: IDF-03 under type IDF stays IDF-03.
1057 lines
36 KiB
Python
1057 lines
36 KiB
Python
"""Network plugin API endpoints."""
|
|
|
|
from flask import Blueprint, request
|
|
from flask_jwt_extended import jwt_required
|
|
|
|
from shopdb.api import db, Asset, AssetType, Vendor, Communication, CommunicationType, AuditLog, success_response, error_response, paginated_response, ErrorCodes, get_pagination_params, paginate_query
|
|
|
|
from ..services.assetnumbers import generate_for_type
|
|
from ..models import NetworkDevice, NetworkDeviceType, Subnet, VLAN
|
|
|
|
from shopdb.api import require_permission, apply_import_timestamps
|
|
|
|
network_bp = Blueprint('network', __name__)
|
|
|
|
|
|
def _primary_ip(assetid):
|
|
"""The asset's primary IP address string, or None. Mirrors the printer
|
|
route: a device's IP lives in a Communication row, not on the extension."""
|
|
comm = (Communication.query.filter_by(assetid=assetid, isprimary=True).first()
|
|
or Communication.query.filter_by(assetid=assetid).first())
|
|
return comm.ipaddress if comm else None
|
|
|
|
|
|
def _upsert_primary_ip(asset, ip):
|
|
"""Create/update/clear the asset's primary IP Communication from an
|
|
ipaddress string. No-op if the IP CommunicationType is not seeded."""
|
|
ip = (ip or '').strip()
|
|
comm = Communication.query.filter_by(assetid=asset.assetid, isprimary=True).first()
|
|
if not ip:
|
|
# Clearing the field clears the row's IP - it used to return early, so
|
|
# an address could be typed but never taken back out.
|
|
if comm:
|
|
comm.ipaddress = None
|
|
return
|
|
if comm:
|
|
comm.ipaddress = ip
|
|
return
|
|
ip_comtype = CommunicationType.query.filter_by(comtype='IP').first()
|
|
if ip_comtype:
|
|
db.session.add(Communication(
|
|
assetid=asset.assetid, comtypeid=ip_comtype.comtypeid,
|
|
ipaddress=ip, isprimary=True))
|
|
|
|
|
|
# =============================================================================
|
|
# Network Device Types
|
|
# =============================================================================
|
|
|
|
@network_bp.route('/types', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_network_device_types():
|
|
"""List all network device types."""
|
|
page, per_page = get_pagination_params(request)
|
|
|
|
query = NetworkDeviceType.query
|
|
|
|
if request.args.get('active', 'true').lower() != 'false':
|
|
query = query.filter(NetworkDeviceType.isactive == True)
|
|
|
|
if search := request.args.get('search'):
|
|
query = query.filter(NetworkDeviceType.networkdevicetype.ilike(f'%{search}%'))
|
|
|
|
query = query.order_by(NetworkDeviceType.networkdevicetype)
|
|
|
|
items, total = paginate_query(query, page, per_page)
|
|
data = [t.to_dict() for t in items]
|
|
|
|
return paginated_response(data, page, per_page, total)
|
|
|
|
|
|
@network_bp.route('/types/<int:type_id>', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def get_network_device_type(type_id: int):
|
|
"""Get a single network device type."""
|
|
t = db.session.get(NetworkDeviceType, type_id)
|
|
|
|
if not t:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Network device type with ID {type_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
return success_response(t.to_dict())
|
|
|
|
|
|
@network_bp.route('/types', methods=['POST'])
|
|
@jwt_required()
|
|
@require_permission('network.create')
|
|
def create_network_device_type():
|
|
"""Create a new network device type."""
|
|
data = request.get_json()
|
|
|
|
if not data or not data.get('networkdevicetype'):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'networkdevicetype is required')
|
|
|
|
existing = NetworkDeviceType.query.filter_by(networkdevicetype=data['networkdevicetype']).first()
|
|
if existing:
|
|
if not existing.isactive:
|
|
existing.isactive = True
|
|
for key in ('description', 'icon', 'color'):
|
|
if data.get(key) is not None:
|
|
setattr(existing, key, data[key])
|
|
db.session.commit()
|
|
return success_response(existing.to_dict(), message='Reactivated existing type')
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Network device type '{data['networkdevicetype']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
t = NetworkDeviceType(
|
|
networkdevicetype=data['networkdevicetype'],
|
|
description=data.get('description'),
|
|
icon=data.get('icon'), color=data.get('color'),
|
|
prefix=data.get('prefix')
|
|
)
|
|
|
|
db.session.add(t)
|
|
db.session.commit()
|
|
|
|
return success_response(t.to_dict(), message='Network device type created', http_code=201)
|
|
|
|
|
|
@network_bp.route('/types/<int:type_id>', methods=['PUT'])
|
|
@jwt_required()
|
|
@require_permission('network.edit')
|
|
def update_network_device_type(type_id: int):
|
|
"""Update a network device type."""
|
|
t = db.session.get(NetworkDeviceType, type_id)
|
|
|
|
if not t:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Network device type with ID {type_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
if 'networkdevicetype' in data and data['networkdevicetype'] != t.networkdevicetype:
|
|
if NetworkDeviceType.query.filter_by(networkdevicetype=data['networkdevicetype']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Network device type '{data['networkdevicetype']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
for key in ['networkdevicetype', 'description', 'icon', 'color', 'prefix',
|
|
'isactive']:
|
|
if key in data:
|
|
setattr(t, key, data[key])
|
|
|
|
db.session.commit()
|
|
return success_response(t.to_dict(), message='Network device type updated')
|
|
|
|
|
|
@network_bp.route('/types/<int:type_id>', methods=['DELETE'])
|
|
@jwt_required()
|
|
@require_permission('network.delete')
|
|
def delete_network_device_type(type_id: int):
|
|
"""Delete a network device type. Refused if any device still uses it."""
|
|
t = db.session.get(NetworkDeviceType, type_id)
|
|
if not t:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Network device type not found', http_code=404)
|
|
inuse = NetworkDevice.query.filter_by(networkdevicetypeid=type_id).count()
|
|
if inuse:
|
|
return error_response(ErrorCodes.CONFLICT,
|
|
f"Cannot delete: {inuse} device(s) still use this type", http_code=409)
|
|
db.session.delete(t)
|
|
db.session.commit()
|
|
return success_response(message='Network device type deleted')
|
|
|
|
|
|
# =============================================================================
|
|
# Network Devices CRUD
|
|
# =============================================================================
|
|
|
|
@network_bp.route('', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_network_devices():
|
|
"""
|
|
List all network devices with filtering and pagination.
|
|
|
|
Query parameters:
|
|
- page, per_page: Pagination
|
|
- active: Filter by active status
|
|
- search: Search by asset number, name, or hostname
|
|
- type_id: Filter by network device type ID
|
|
- vendor_id: Filter by vendor ID
|
|
- location_id: Filter by location ID
|
|
- businessunit_id: Filter by business unit ID
|
|
- poe: Filter by PoE capability (true/false)
|
|
- managed: Filter by managed status (true/false)
|
|
"""
|
|
page, per_page = get_pagination_params(request)
|
|
|
|
# Join NetworkDevice with Asset
|
|
query = db.session.query(NetworkDevice).join(Asset)
|
|
|
|
# Active filter
|
|
if request.args.get('active', 'true').lower() != 'false':
|
|
query = query.filter(Asset.isactive == True)
|
|
|
|
# Exact-match natural-key lookup for idempotent import (asset number).
|
|
if exactassetnumber := request.args.get('assetnumber'):
|
|
query = query.filter(Asset.assetnumber == exactassetnumber)
|
|
|
|
# Search filter. Type and vendor are columns in the list, so both must be
|
|
# searchable. Outer joins so a device missing either still matches on its
|
|
# own fields.
|
|
if search := request.args.get('search'):
|
|
pattern = f'%{search}%'
|
|
query = query.outerjoin(
|
|
NetworkDeviceType,
|
|
NetworkDevice.networkdevicetypeid == NetworkDeviceType.networkdevicetypeid
|
|
).outerjoin(
|
|
Vendor, NetworkDevice.vendorid == Vendor.vendorid
|
|
).filter(
|
|
db.or_(
|
|
Asset.assetnumber.ilike(pattern),
|
|
Asset.name.ilike(pattern),
|
|
Asset.serialnumber.ilike(pattern),
|
|
NetworkDevice.hostname.ilike(pattern),
|
|
NetworkDeviceType.networkdevicetype.ilike(pattern),
|
|
Vendor.vendor.ilike(pattern)
|
|
)
|
|
)
|
|
|
|
# Type filter
|
|
if type_id := request.args.get('typeid', request.args.get('type_id')):
|
|
query = query.filter(NetworkDevice.networkdevicetypeid == int(type_id))
|
|
|
|
# Vendor filter
|
|
if vendor_id := request.args.get('vendorid', request.args.get('vendor_id')):
|
|
query = query.filter(NetworkDevice.vendorid == int(vendor_id))
|
|
|
|
# Location filter
|
|
if location_id := request.args.get('locationid', request.args.get('location_id')):
|
|
query = query.filter(Asset.locationid == int(location_id))
|
|
|
|
# Business unit filter
|
|
if bu_id := request.args.get('businessunitid', request.args.get('businessunit_id')):
|
|
query = query.filter(Asset.businessunitid == int(bu_id))
|
|
|
|
# PoE filter
|
|
if poe := request.args.get('poe'):
|
|
query = query.filter(NetworkDevice.ispoe == (poe.lower() == 'true'))
|
|
|
|
# Managed filter
|
|
if managed := request.args.get('managed'):
|
|
query = query.filter(NetworkDevice.ismanaged == (managed.lower() == 'true'))
|
|
|
|
# Sorting
|
|
sort_by = request.args.get('sort', 'hostname')
|
|
sort_dir = request.args.get('dir', 'asc')
|
|
|
|
if sort_by == 'hostname':
|
|
col = NetworkDevice.hostname
|
|
elif sort_by == 'assetnumber':
|
|
col = Asset.assetnumber
|
|
elif sort_by == 'name':
|
|
col = Asset.name
|
|
else:
|
|
col = NetworkDevice.hostname
|
|
|
|
query = query.order_by(col.desc() if sort_dir == 'desc' else col)
|
|
|
|
items, total = paginate_query(query, page, per_page)
|
|
|
|
# Build response with both asset and network device data
|
|
data = []
|
|
for netdev in items:
|
|
item = netdev.asset.to_dict() if netdev.asset else {}
|
|
item['networkdevice'] = netdev.to_dict()
|
|
item['ipaddress'] = _primary_ip(netdev.assetid)
|
|
data.append(item)
|
|
|
|
return paginated_response(data, page, per_page, total)
|
|
|
|
|
|
@network_bp.route('/<int:device_id>', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def get_network_device(device_id: int):
|
|
"""Get a single network device with full details."""
|
|
netdev = db.session.get(NetworkDevice, device_id)
|
|
|
|
if not netdev:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Network device with ID {device_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
result = netdev.asset.to_dict() if netdev.asset else {}
|
|
result['networkdevice'] = netdev.to_dict()
|
|
result['ipaddress'] = _primary_ip(netdev.assetid)
|
|
|
|
return success_response(result)
|
|
|
|
|
|
@network_bp.route('/by-asset/<int:asset_id>', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def get_network_device_by_asset(asset_id: int):
|
|
"""Get network device data by asset ID."""
|
|
netdev = NetworkDevice.query.filter_by(assetid=asset_id).first()
|
|
|
|
if not netdev:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Network device for asset {asset_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
result = netdev.asset.to_dict() if netdev.asset else {}
|
|
result['networkdevice'] = netdev.to_dict()
|
|
result['ipaddress'] = _primary_ip(netdev.assetid)
|
|
|
|
return success_response(result)
|
|
|
|
|
|
@network_bp.route('/by-hostname/<hostname>', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def get_network_device_by_hostname(hostname: str):
|
|
"""Get network device by hostname."""
|
|
netdev = NetworkDevice.query.filter_by(hostname=hostname).first()
|
|
|
|
if not netdev:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Network device with hostname {hostname} not found',
|
|
http_code=404
|
|
)
|
|
|
|
result = netdev.asset.to_dict() if netdev.asset else {}
|
|
result['networkdevice'] = netdev.to_dict()
|
|
result['ipaddress'] = _primary_ip(netdev.assetid)
|
|
|
|
return success_response(result)
|
|
|
|
|
|
@network_bp.route('', methods=['POST'])
|
|
@jwt_required()
|
|
@require_permission('network.create')
|
|
def create_network_device():
|
|
"""
|
|
Create new network device (creates both Asset and NetworkDevice records).
|
|
|
|
Required fields:
|
|
- assetnumber: Business identifier
|
|
|
|
Optional fields:
|
|
- name, serialnumber, statusid, locationid, businessunitid
|
|
- networkdevicetypeid, vendorid, modelnumberid, hostname
|
|
- firmwareversion, portcount, ispoe, ismanaged, rackunit
|
|
- mapx, mapy, notes
|
|
"""
|
|
data = request.get_json()
|
|
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
# Generate from the type prefix + name when left blank. Only when blank: a
|
|
# device carrying a real identifier of its own - vendor tag, controller
|
|
# name, serial - keeps it, which is the platform rule.
|
|
if not data.get('assetnumber'):
|
|
data['assetnumber'] = generate_for_type(
|
|
data.get('networkdevicetypeid'), data.get('name'))
|
|
if not data.get('assetnumber'):
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'assetnumber is required (or give a name, and a device type with a '
|
|
'prefix, to have one generated)')
|
|
|
|
# Check for duplicate assetnumber
|
|
if Asset.query.filter_by(assetnumber=data['assetnumber']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Asset with number '{data['assetnumber']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
# Check for duplicate hostname
|
|
if data.get('hostname'):
|
|
if NetworkDevice.query.filter_by(hostname=data['hostname']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Network device with hostname '{data['hostname']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
# Get network device asset type
|
|
network_type = AssetType.query.filter_by(assettype='network_device').first()
|
|
if not network_type:
|
|
return error_response(
|
|
ErrorCodes.INTERNAL_ERROR,
|
|
'Network device asset type not found. Plugin may not be properly installed.',
|
|
http_code=500
|
|
)
|
|
|
|
# Create the core asset
|
|
asset = Asset(
|
|
assetnumber=data['assetnumber'],
|
|
name=data.get('name'),
|
|
serialnumber=data.get('serialnumber'),
|
|
gaugelabreference=data.get('gaugelabreference'),
|
|
maintenancereference=data.get('maintenancereference'),
|
|
assettypeid=network_type.assettypeid,
|
|
statusid=data.get('statusid', 1),
|
|
locationid=data.get('locationid'),
|
|
businessunitid=data.get('businessunitid'),
|
|
mapx=data.get('mapx'),
|
|
mapy=data.get('mapy'),
|
|
notes=data.get('notes')
|
|
)
|
|
|
|
db.session.add(asset)
|
|
db.session.flush() # Get the assetid
|
|
|
|
# Create the network device extension
|
|
netdev = NetworkDevice(
|
|
assetid=asset.assetid,
|
|
networkdevicetypeid=data.get('networkdevicetypeid'),
|
|
vendorid=data.get('vendorid'),
|
|
modelnumberid=data.get('modelnumberid'),
|
|
hostname=data.get('hostname'),
|
|
firmwareversion=data.get('firmwareversion'),
|
|
portcount=data.get('portcount'),
|
|
ispoe=data.get('ispoe', False),
|
|
ismanaged=data.get('ismanaged', False),
|
|
rackunit=data.get('rackunit')
|
|
)
|
|
|
|
db.session.add(netdev)
|
|
db.session.flush()
|
|
|
|
# Attach the primary IP as a Communication when supplied (a device's IP is
|
|
# not a NetworkDevice column). Mirrors the printer create route.
|
|
if data.get('ipaddress'):
|
|
_upsert_primary_ip(asset, data['ipaddress'])
|
|
|
|
# Preserve legacy timestamps in import mode (no-op otherwise)
|
|
apply_import_timestamps(asset, data)
|
|
|
|
# Audit log
|
|
AuditLog.log('created', 'NetworkDevice', entityid=netdev.networkdeviceid,
|
|
entityname=data.get('hostname') or data['assetnumber'])
|
|
|
|
db.session.commit()
|
|
|
|
result = asset.to_dict()
|
|
result['networkdevice'] = netdev.to_dict()
|
|
result['ipaddress'] = _primary_ip(asset.assetid)
|
|
|
|
return success_response(result, message='Network device created', http_code=201)
|
|
|
|
|
|
@network_bp.route('/<int:device_id>', methods=['PUT'])
|
|
@jwt_required()
|
|
@require_permission('network.edit')
|
|
def update_network_device(device_id: int):
|
|
"""Update network device (both Asset and NetworkDevice records)."""
|
|
netdev = db.session.get(NetworkDevice, device_id)
|
|
|
|
if not netdev:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Network device with ID {device_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
asset = netdev.asset
|
|
|
|
# Check for conflicting assetnumber
|
|
if 'assetnumber' in data and data['assetnumber'] != asset.assetnumber:
|
|
if Asset.query.filter_by(assetnumber=data['assetnumber']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Asset with number '{data['assetnumber']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
# Check for conflicting hostname
|
|
if 'hostname' in data and data['hostname'] != netdev.hostname:
|
|
existing = NetworkDevice.query.filter_by(hostname=data['hostname']).first()
|
|
if existing and existing.networkdeviceid != device_id:
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Network device with hostname '{data['hostname']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
# Track changes for audit log
|
|
changes = {}
|
|
|
|
# Update asset fields
|
|
asset_fields = ['assetnumber', 'name', 'serialnumber', 'gaugelabreference',
|
|
'maintenancereference', 'statusid',
|
|
'locationid', 'businessunitid', 'mapx', 'mapy', 'notes', 'isactive']
|
|
for key in asset_fields:
|
|
if key in data:
|
|
old_val = getattr(asset, key)
|
|
new_val = data[key]
|
|
if old_val != new_val:
|
|
changes[key] = {'old': old_val, 'new': new_val}
|
|
setattr(asset, key, data[key])
|
|
|
|
# Update network device fields
|
|
netdev_fields = ['networkdevicetypeid', 'vendorid', 'modelnumberid', 'hostname',
|
|
'firmwareversion', 'portcount', 'ispoe', 'ismanaged', 'rackunit']
|
|
for key in netdev_fields:
|
|
if key in data:
|
|
old_val = getattr(netdev, key)
|
|
new_val = data[key]
|
|
if old_val != new_val:
|
|
changes[key] = {'old': old_val, 'new': new_val}
|
|
setattr(netdev, key, data[key])
|
|
|
|
# Audit log if there were changes
|
|
if changes:
|
|
AuditLog.log('updated', 'NetworkDevice', entityid=netdev.networkdeviceid,
|
|
entityname=netdev.hostname or asset.assetnumber, changes=changes)
|
|
|
|
# Upsert the primary IP Communication when an ipaddress is supplied.
|
|
if 'ipaddress' in data:
|
|
_upsert_primary_ip(asset, data['ipaddress'])
|
|
|
|
apply_import_timestamps(asset, data)
|
|
db.session.commit()
|
|
|
|
result = asset.to_dict()
|
|
result['networkdevice'] = netdev.to_dict()
|
|
result['ipaddress'] = _primary_ip(asset.assetid)
|
|
|
|
return success_response(result, message='Network device updated')
|
|
|
|
|
|
@network_bp.route('/<int:device_id>', methods=['DELETE'])
|
|
@jwt_required()
|
|
@require_permission('network.delete')
|
|
def delete_network_device(device_id: int):
|
|
"""Delete (soft delete) network device."""
|
|
netdev = db.session.get(NetworkDevice, device_id)
|
|
|
|
if not netdev:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Network device with ID {device_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
# Soft delete the asset
|
|
netdev.asset.isactive = False
|
|
|
|
# Audit log
|
|
AuditLog.log('deleted', 'NetworkDevice', entityid=netdev.networkdeviceid,
|
|
entityname=netdev.hostname or netdev.asset.assetnumber)
|
|
|
|
db.session.commit()
|
|
|
|
return success_response(message='Network device deleted')
|
|
|
|
|
|
# =============================================================================
|
|
# Dashboard
|
|
# =============================================================================
|
|
|
|
@network_bp.route('/dashboard/summary', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def dashboard_summary():
|
|
"""Get network device dashboard summary data."""
|
|
# Total active network devices
|
|
total = db.session.query(NetworkDevice).join(Asset).filter(
|
|
Asset.isactive == True
|
|
).count()
|
|
|
|
# Count by device type
|
|
by_type = db.session.query(
|
|
NetworkDeviceType.networkdevicetype,
|
|
db.func.count(NetworkDevice.networkdeviceid)
|
|
).join(NetworkDevice, NetworkDevice.networkdevicetypeid == NetworkDeviceType.networkdevicetypeid
|
|
).join(Asset, Asset.assetid == NetworkDevice.assetid
|
|
).filter(Asset.isactive == True
|
|
).group_by(NetworkDeviceType.networkdevicetype
|
|
).all()
|
|
|
|
# Count by vendor
|
|
by_vendor = db.session.query(
|
|
Vendor.vendor,
|
|
db.func.count(NetworkDevice.networkdeviceid)
|
|
).join(NetworkDevice, NetworkDevice.vendorid == Vendor.vendorid
|
|
).join(Asset, Asset.assetid == NetworkDevice.assetid
|
|
).filter(Asset.isactive == True
|
|
).group_by(Vendor.vendor
|
|
).all()
|
|
|
|
# Count PoE vs non-PoE
|
|
poe_count = db.session.query(NetworkDevice).join(Asset).filter(
|
|
Asset.isactive == True,
|
|
NetworkDevice.ispoe == True
|
|
).count()
|
|
|
|
return success_response({
|
|
'total': total,
|
|
'bytype': [{'type': t, 'count': c} for t, c in by_type],
|
|
'byvendor': [{'vendor': v, 'count': c} for v, c in by_vendor],
|
|
'poe': poe_count,
|
|
'nonpoe': total - poe_count
|
|
})
|
|
|
|
|
|
# =============================================================================
|
|
# VLANs
|
|
# =============================================================================
|
|
|
|
@network_bp.route('/vlans', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_vlans():
|
|
"""List all VLANs with filtering and pagination."""
|
|
page, per_page = get_pagination_params(request)
|
|
|
|
query = VLAN.query
|
|
|
|
if request.args.get('active', 'true').lower() != 'false':
|
|
query = query.filter(VLAN.isactive == True)
|
|
|
|
# Search filter
|
|
if search := request.args.get('search'):
|
|
query = query.filter(
|
|
db.or_(
|
|
VLAN.name.ilike(f'%{search}%'),
|
|
VLAN.description.ilike(f'%{search}%'),
|
|
db.cast(VLAN.vlannumber, db.String).ilike(f'%{search}%')
|
|
)
|
|
)
|
|
|
|
# Type filter
|
|
if vlan_type := request.args.get('type'):
|
|
query = query.filter(VLAN.vlantype == vlan_type)
|
|
|
|
query = query.order_by(VLAN.vlannumber)
|
|
|
|
items, total = paginate_query(query, page, per_page)
|
|
data = [v.to_dict() for v in items]
|
|
|
|
return paginated_response(data, page, per_page, total)
|
|
|
|
|
|
@network_bp.route('/vlans/<int:vlan_id>', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def get_vlan(vlan_id: int):
|
|
"""Get a single VLAN with its subnets."""
|
|
vlan = db.session.get(VLAN, vlan_id)
|
|
|
|
if not vlan:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'VLAN with ID {vlan_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
result = vlan.to_dict()
|
|
# Include associated subnets
|
|
result['subnets'] = [s.to_dict() for s in vlan.subnets.filter_by(isactive=True).all()]
|
|
|
|
return success_response(result)
|
|
|
|
|
|
@network_bp.route('/vlans', methods=['POST'])
|
|
@jwt_required()
|
|
@require_permission('network.create')
|
|
def create_vlan():
|
|
"""Create a new VLAN."""
|
|
data = request.get_json()
|
|
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
if not data.get('vlannumber'):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'vlannumber is required')
|
|
if not data.get('name'):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'name is required')
|
|
|
|
# Check for duplicate VLAN number
|
|
if VLAN.query.filter_by(vlannumber=data['vlannumber']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"VLAN {data['vlannumber']} already exists",
|
|
http_code=409
|
|
)
|
|
|
|
vlan = VLAN(
|
|
vlannumber=data['vlannumber'],
|
|
name=data['name'],
|
|
description=data.get('description'),
|
|
vlantype=data.get('vlantype')
|
|
)
|
|
|
|
db.session.add(vlan)
|
|
db.session.flush()
|
|
|
|
# Audit log
|
|
AuditLog.log('created', 'VLAN', entityid=vlan.vlanid,
|
|
entityname=f"VLAN {vlan.vlannumber} - {vlan.name}")
|
|
|
|
db.session.commit()
|
|
|
|
return success_response(vlan.to_dict(), message='VLAN created', http_code=201)
|
|
|
|
|
|
@network_bp.route('/vlans/<int:vlan_id>', methods=['PUT'])
|
|
@jwt_required()
|
|
@require_permission('network.edit')
|
|
def update_vlan(vlan_id: int):
|
|
"""Update a VLAN."""
|
|
vlan = db.session.get(VLAN, vlan_id)
|
|
|
|
if not vlan:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'VLAN with ID {vlan_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
# Check for conflicting VLAN number
|
|
if 'vlannumber' in data and data['vlannumber'] != vlan.vlannumber:
|
|
if VLAN.query.filter_by(vlannumber=data['vlannumber']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"VLAN {data['vlannumber']} already exists",
|
|
http_code=409
|
|
)
|
|
|
|
# Track changes for audit log
|
|
changes = {}
|
|
for key in ['vlannumber', 'name', 'description', 'vlantype', 'isactive']:
|
|
if key in data:
|
|
old_val = getattr(vlan, key)
|
|
new_val = data[key]
|
|
if old_val != new_val:
|
|
changes[key] = {'old': old_val, 'new': new_val}
|
|
setattr(vlan, key, data[key])
|
|
|
|
# Audit log if there were changes
|
|
if changes:
|
|
AuditLog.log('updated', 'VLAN', entityid=vlan.vlanid,
|
|
entityname=f"VLAN {vlan.vlannumber} - {vlan.name}", changes=changes)
|
|
|
|
db.session.commit()
|
|
return success_response(vlan.to_dict(), message='VLAN updated')
|
|
|
|
|
|
@network_bp.route('/vlans/<int:vlan_id>', methods=['DELETE'])
|
|
@jwt_required()
|
|
@require_permission('network.delete')
|
|
def delete_vlan(vlan_id: int):
|
|
"""Delete (soft delete) a VLAN."""
|
|
vlan = db.session.get(VLAN, vlan_id)
|
|
|
|
if not vlan:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'VLAN with ID {vlan_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
# Check if VLAN has associated subnets
|
|
if vlan.subnets.filter_by(isactive=True).count() > 0:
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'Cannot delete VLAN with associated subnets',
|
|
http_code=400
|
|
)
|
|
|
|
vlan.isactive = False
|
|
|
|
# Audit log
|
|
AuditLog.log('deleted', 'VLAN', entityid=vlan.vlanid,
|
|
entityname=f"VLAN {vlan.vlannumber} - {vlan.name}")
|
|
|
|
db.session.commit()
|
|
|
|
return success_response(message='VLAN deleted')
|
|
|
|
|
|
# =============================================================================
|
|
# Subnets
|
|
# =============================================================================
|
|
|
|
@network_bp.route('/subnets', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_subnets():
|
|
"""List all subnets with filtering and pagination."""
|
|
page, per_page = get_pagination_params(request)
|
|
|
|
query = Subnet.query
|
|
|
|
if request.args.get('active', 'true').lower() != 'false':
|
|
query = query.filter(Subnet.isactive == True)
|
|
|
|
# Search filter
|
|
if search := request.args.get('search'):
|
|
query = query.filter(
|
|
db.or_(
|
|
Subnet.cidr.ilike(f'%{search}%'),
|
|
Subnet.name.ilike(f'%{search}%'),
|
|
Subnet.description.ilike(f'%{search}%')
|
|
)
|
|
)
|
|
|
|
# VLAN filter
|
|
if vlan_id := request.args.get('vlanid'):
|
|
query = query.filter(Subnet.vlanid == int(vlan_id))
|
|
|
|
# Location filter
|
|
if location_id := request.args.get('locationid'):
|
|
query = query.filter(Subnet.locationid == int(location_id))
|
|
|
|
# Type filter
|
|
if subnet_type := request.args.get('type'):
|
|
query = query.filter(Subnet.subnettype == subnet_type)
|
|
|
|
query = query.order_by(Subnet.cidr)
|
|
|
|
items, total = paginate_query(query, page, per_page)
|
|
data = [s.to_dict() for s in items]
|
|
|
|
return paginated_response(data, page, per_page, total)
|
|
|
|
|
|
@network_bp.route('/subnets/<int:subnet_id>', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def get_subnet(subnet_id: int):
|
|
"""Get a single subnet plus the network devices whose primary IP falls in
|
|
its CIDR range."""
|
|
subnet = db.session.get(Subnet, subnet_id)
|
|
|
|
if not subnet:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Subnet with ID {subnet_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
data = subnet.to_dict()
|
|
data['devices'] = _devices_in_subnet(subnet.cidr)
|
|
return success_response(data)
|
|
|
|
|
|
def _devices_in_subnet(cidr):
|
|
"""ANY asset (PC, printer, network device, ...) whose primary IP falls in
|
|
cidr. A subnet is cross-type, so this matches on the core Communication +
|
|
Asset tables, not just network devices. Matching is in Python because the IP
|
|
lives in a Communication row, not a column."""
|
|
import ipaddress
|
|
from shopdb.api import Asset, AssetType, Communication
|
|
try:
|
|
net = ipaddress.ip_network(cidr, strict=False)
|
|
except (ValueError, TypeError):
|
|
return []
|
|
rows = (Communication.query
|
|
.join(Asset, Communication.assetid == Asset.assetid)
|
|
.join(AssetType, Asset.assettypeid == AssetType.assettypeid)
|
|
.filter(Communication.isprimary == True,
|
|
Communication.ipaddress.isnot(None))
|
|
.with_entities(Asset.assetid, Asset.assetnumber, Asset.name,
|
|
AssetType.assettype, Communication.ipaddress).all())
|
|
result = []
|
|
for assetid, assetnumber, name, assettype, ip in rows:
|
|
ip = (ip or '').strip()
|
|
if not ip:
|
|
continue
|
|
try:
|
|
if ipaddress.ip_address(ip) in net:
|
|
result.append({
|
|
'assetid': assetid, 'assetnumber': assetnumber, 'name': name,
|
|
'assettype': assettype, 'ipaddress': ip,
|
|
'url': _asset_detail_url(assettype, assetid),
|
|
})
|
|
except (ValueError, TypeError):
|
|
continue
|
|
result.sort(key=lambda d: d['ipaddress'])
|
|
return result
|
|
|
|
|
|
def _asset_detail_url(assettype, assetid):
|
|
"""Best-effort front-end detail path for an asset by type. Resolves the
|
|
per-type extension id lazily/guarded (subnet listings span plugins); returns
|
|
None when the plugin is absent so the frontend just shows the row."""
|
|
try:
|
|
if assettype == 'network_device':
|
|
row = NetworkDevice.query.filter_by(assetid=assetid).first()
|
|
return f'/network/{row.networkdeviceid}' if row else None
|
|
if assettype == 'printer':
|
|
from plugins.printers.models import Printer
|
|
row = Printer.query.filter_by(assetid=assetid).first()
|
|
return f'/printers/{row.printerid}' if row else None
|
|
if assettype == 'computer':
|
|
from plugins.computers.models import Computer
|
|
row = Computer.query.filter_by(assetid=assetid).first()
|
|
return f'/pcs/{row.computerid}' if row else None
|
|
except ImportError:
|
|
return None
|
|
return None
|
|
|
|
|
|
@network_bp.route('/subnets', methods=['POST'])
|
|
@jwt_required()
|
|
@require_permission('network.create')
|
|
def create_subnet():
|
|
"""Create a new subnet."""
|
|
data = request.get_json()
|
|
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
if not data.get('cidr'):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'cidr is required')
|
|
if not data.get('name'):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'name is required')
|
|
|
|
# Validate CIDR format (basic check)
|
|
cidr = data['cidr']
|
|
if '/' not in cidr:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'cidr must be in CIDR notation (e.g., 10.1.1.0/24)')
|
|
|
|
# Check for duplicate CIDR
|
|
if Subnet.query.filter_by(cidr=cidr).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Subnet {cidr} already exists",
|
|
http_code=409
|
|
)
|
|
|
|
# Validate VLAN if provided
|
|
if data.get('vlanid'):
|
|
if not db.session.get(VLAN, data['vlanid']):
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
f"VLAN with ID {data['vlanid']} not found"
|
|
)
|
|
|
|
subnet = Subnet(
|
|
cidr=cidr,
|
|
name=data['name'],
|
|
description=data.get('description'),
|
|
gatewayip=data.get('gatewayip'),
|
|
subnetmask=data.get('subnetmask'),
|
|
networkaddress=data.get('networkaddress'),
|
|
broadcastaddress=data.get('broadcastaddress'),
|
|
vlanid=data.get('vlanid'),
|
|
subnettype=data.get('subnettype'),
|
|
locationid=data.get('locationid'),
|
|
dhcpenabled=data.get('dhcpenabled', True),
|
|
dhcprangestart=data.get('dhcprangestart'),
|
|
dhcprangeend=data.get('dhcprangeend'),
|
|
dns1=data.get('dns1'),
|
|
dns2=data.get('dns2')
|
|
)
|
|
|
|
db.session.add(subnet)
|
|
db.session.flush()
|
|
|
|
# Audit log
|
|
AuditLog.log('created', 'Subnet', entityid=subnet.subnetid,
|
|
entityname=f"{subnet.cidr} - {subnet.name}")
|
|
|
|
db.session.commit()
|
|
|
|
return success_response(subnet.to_dict(), message='Subnet created', http_code=201)
|
|
|
|
|
|
@network_bp.route('/subnets/<int:subnet_id>', methods=['PUT'])
|
|
@jwt_required()
|
|
@require_permission('network.edit')
|
|
def update_subnet(subnet_id: int):
|
|
"""Update a subnet."""
|
|
subnet = db.session.get(Subnet, subnet_id)
|
|
|
|
if not subnet:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Subnet with ID {subnet_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
# Check for conflicting CIDR
|
|
if 'cidr' in data and data['cidr'] != subnet.cidr:
|
|
if Subnet.query.filter_by(cidr=data['cidr']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Subnet {data['cidr']} already exists",
|
|
http_code=409
|
|
)
|
|
|
|
allowed_fields = ['cidr', 'name', 'description', 'gatewayip', 'subnetmask',
|
|
'networkaddress', 'broadcastaddress', 'vlanid', 'subnettype',
|
|
'locationid', 'dhcpenabled', 'dhcprangestart', 'dhcprangeend',
|
|
'dns1', 'dns2', 'isactive']
|
|
|
|
# Track changes for audit log
|
|
changes = {}
|
|
for key in allowed_fields:
|
|
if key in data:
|
|
old_val = getattr(subnet, key)
|
|
new_val = data[key]
|
|
if old_val != new_val:
|
|
changes[key] = {'old': old_val, 'new': new_val}
|
|
setattr(subnet, key, data[key])
|
|
|
|
# Audit log if there were changes
|
|
if changes:
|
|
AuditLog.log('updated', 'Subnet', entityid=subnet.subnetid,
|
|
entityname=f"{subnet.cidr} - {subnet.name}", changes=changes)
|
|
|
|
db.session.commit()
|
|
return success_response(subnet.to_dict(), message='Subnet updated')
|
|
|
|
|
|
@network_bp.route('/subnets/<int:subnet_id>', methods=['DELETE'])
|
|
@jwt_required()
|
|
@require_permission('network.delete')
|
|
def delete_subnet(subnet_id: int):
|
|
"""Delete (soft delete) a subnet."""
|
|
subnet = db.session.get(Subnet, subnet_id)
|
|
|
|
if not subnet:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Subnet with ID {subnet_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
subnet.isactive = False
|
|
|
|
# Audit log
|
|
AuditLog.log('deleted', 'Subnet', entityid=subnet.subnetid,
|
|
entityname=f"{subnet.cidr} - {subnet.name}")
|
|
|
|
db.session.commit()
|
|
|
|
return success_response(message='Subnet deleted')
|