"""Polymorphic Asset models - core of the new asset architecture.""" from shopdb.extensions import db from .base import BaseModel, SoftDeleteMixin, AuditMixin class AssetType(BaseModel): """ Registry of asset categories. Each type maps to a plugin-owned extension table. Examples: equipment, computer, network_device, printer """ __tablename__ = 'assettypes' assettypeid = db.Column(db.Integer, primary_key=True) assettype = db.Column( db.String(50), unique=True, nullable=False, comment='Category name: equipment, computer, network_device, printer' ) pluginname = db.Column( db.String(100), nullable=True, comment='Plugin that owns this type' ) tablename = db.Column( db.String(100), nullable=True, comment='Extension table name for this type' ) description = db.Column(db.Text) icon = db.Column(db.String(50), comment='Icon name for UI') def __repr__(self): return f"" class AssetStatus(BaseModel): """Asset status options.""" __tablename__ = 'assetstatuses' statusid = db.Column(db.Integer, primary_key=True) status = db.Column(db.String(50), unique=True, nullable=False) description = db.Column(db.Text) color = db.Column(db.String(20), comment='CSS color for UI') def __repr__(self): return f"" class Asset(BaseModel, SoftDeleteMixin, AuditMixin): """ Core asset model - minimal shared fields. Category-specific data lives in plugin extension tables (equipment, computers, network_devices, printers). The assetid matches original machineid for migration compatibility. """ __tablename__ = 'assets' assetid = db.Column(db.Integer, primary_key=True) # Identification assetnumber = db.Column( db.String(50), unique=True, nullable=False, index=True, comment='Business identifier (e.g., CMM01, G5QX1GT3ESF)' ) name = db.Column( db.String(100), comment='Display name/alias' ) serialnumber = db.Column( db.String(100), index=True, comment='Hardware serial number' ) # Classification assettypeid = db.Column( db.Integer, db.ForeignKey('assettypes.assettypeid'), nullable=False ) statusid = db.Column( db.Integer, db.ForeignKey('assetstatuses.statusid'), default=1, comment='In Use, Spare, Retired, etc.' ) # Location and organization locationid = db.Column( db.Integer, db.ForeignKey('locations.locationid'), nullable=True ) businessunitid = db.Column( db.Integer, db.ForeignKey('businessunits.businessunitid'), nullable=True ) # Floor map position (ADR-001: asset-specific override; nullable) mapx = db.Column(db.Integer, comment='X coordinate on floor map (ADR-001)') mapy = db.Column(db.Integer, comment='Y coordinate on floor map (ADR-001)') # Notes notes = db.Column(db.Text, nullable=True) # Relationships assettype = db.relationship('AssetType', backref='assets') status = db.relationship('AssetStatus', backref='assets') location = db.relationship('Location', backref='assets') businessunit = db.relationship('BusinessUnit', backref='assets') # Communications (one-to-many) - will be migrated to use assetid communications = db.relationship( 'Communication', foreign_keys='Communication.assetid', backref='asset', cascade='all, delete-orphan', lazy='dynamic' ) # Indexes __table_args__ = ( db.Index('idx_asset_type_bu', 'assettypeid', 'businessunitid'), db.Index('idx_asset_location', 'locationid'), db.Index('idx_asset_active', 'isactive'), db.Index('idx_asset_status', 'statusid'), ) def __repr__(self): return f"" @property def display_name(self): """Get display name (name if set, otherwise assetnumber).""" return self.name or self.assetnumber @property def primary_ip(self): """Get primary IP address from communications.""" comm = self.communications.filter_by( isprimary=True, comtypeid=1 # IP type ).first() if comm: return comm.ipaddress # Fall back to any IP comm = self.communications.filter_by(comtypeid=1).first() return comm.ipaddress if comm else None def get_inherited_location(self): """ Get location data from a related asset if this asset has none. Returns dict with locationid, location_name, mapx, mapy, and inherited_from (assetnumber of source asset) if location was inherited. Returns None if no location data available. """ if self.locationid is not None or (self.mapx is not None and self.mapy is not None): return None related_assets = [] if hasattr(self, 'incoming_relationships'): for rel in self.incoming_relationships: if rel.sourceasset and rel.isactive: related_assets.append(rel.sourceasset) if hasattr(self, 'outgoing_relationships'): for rel in self.outgoing_relationships: if rel.targetasset and rel.isactive: related_assets.append(rel.targetasset) for related in related_assets: if related.locationid is not None or (related.mapx is not None and related.mapy is not None): return { 'locationid': related.locationid, 'locationname': related.location.locationname if related.location else None, 'mapx': related.mapx, 'mapy': related.mapy, 'inheritedfrom': related.assetnumber } return None def to_dict(self, include_type_data=False, include_inherited_location=True): """ Convert model to dictionary. Args: include_type_data: If True, include category-specific data from extension table include_inherited_location: If True, include location from related assets when missing """ result = super().to_dict() # Add related object names for convenience if self.assettype: result['assettypename'] = self.assettype.assettype if self.status: result['statusname'] = self.status.status if self.location: result['locationname'] = self.location.locationname if self.businessunit: result['businessunitname'] = self.businessunit.businessunit # Add plugin-specific ID for navigation purposes if hasattr(self, 'equipment') and self.equipment: result['pluginid'] = self.equipment.equipmentid elif hasattr(self, 'computer') and self.computer: result['pluginid'] = self.computer.computerid elif hasattr(self, 'network_device') and self.network_device: result['pluginid'] = self.network_device.networkdeviceid elif hasattr(self, 'printer') and self.printer: result['pluginid'] = self.printer.printerid # Include inherited location if this asset has no location data if include_inherited_location: inherited = self.get_inherited_location() if inherited: result['inheritedlocation'] = inherited # Also set the location fields if they're missing if result.get('locationid') is None: result['locationid'] = inherited['locationid'] result['locationname'] = inherited['locationname'] if result.get('mapx') is None: result['mapx'] = inherited['mapx'] if result.get('mapy') is None: result['mapy'] = inherited['mapy'] # Include extension data if requested if include_type_data: ext_data = self._get_extension_data() if ext_data: result['typedata'] = ext_data return result def _get_extension_data(self): """Get category-specific data from extension table.""" # Check for equipment extension if hasattr(self, 'equipment') and self.equipment: return self.equipment.to_dict() # Check for computer extension if hasattr(self, 'computer') and self.computer: return self.computer.to_dict() # Check for network_device extension if hasattr(self, 'network_device') and self.network_device: return self.network_device.to_dict() # Check for printer extension if hasattr(self, 'printer') and self.printer: return self.printer.to_dict() return None