Files
shopdb-flask/plugins/computers/models/computer.py
cproudlock 1c6c7ba14b
Some checks failed
CI / backend (push) Successful in 1m39s
CI / naming (push) Successful in 1s
CI / frontend (push) Successful in 8s
CI / migrations-mysql (push) Failing after 8s
DB review fixes: drop redundant indexes + dead column, add CI MySQL-upgrade job
From the database review (verdict: sound-with-minor-issues). Applies the
actionable findings.

Redundant indexes: five non-unique secondary indexes duplicated a named idx_*
or a unique index on the same column - ix_communications_assetid,
ix_computers_hostname, ix_networkdevices_hostname, ix_printers_hostname (each
shadowing an idx_*), and idx_usb_serial (shadowing the serialnumber unique
index). Removed the redundant index source from the models (column index=True /
the extra db.Index) and added core migration 7d25 dropping the live duplicates.
The unique ix_*_assetid indexes are kept (they enforce assetid uniqueness).

Dead column: usbcheckouts.machineid was a NOT NULL soft-ref to the retired
machines table storing sentinel 0 (ADR-001). Dropped from the model + the
machineid=0 literal in selfhosted checkout; usb plugin migration 0002 drops it
live (downgrade restores it default 0).

Index: notifications.businessunitid (filtered by the shopfloor feed) was
unindexed; added index=True + notifications migration 0002.

CI: new migrations-mysql job proves the real multi-site deploy path - fresh
`flask db upgrade` + per-plugin install on utf8mb4 MySQL from empty, asserting
table count + charset and a clean second-run no-op. The pytest suite only
exercises SQLite create_all(), so a regression in the Alembic chain on MySQL
would otherwise ship undetected.

Verified: fresh core upgrade on a scratch utf8mb4 MySQL builds clean + no-op on
rerun (redundant indexes absent, unique assetid kept); plugin migrations applied
+ verified on the dev DB (machineid gone, bu index present). 953 backend tests
pass; naming + pyflakes green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-13 09:29:45 -04:00

285 lines
9.1 KiB
Python

"""Computer plugin models."""
from datetime import datetime, timezone
from shopdb.api import db, BaseModel
def _utcnow():
# naive UTC to match the other DB DateTime columns (stored without tzinfo)
return datetime.now(timezone.utc).replace(tzinfo=None)
class ComputerType(BaseModel):
"""
Computer type classification.
Examples: Shopfloor PC, Engineer Workstation, CMM PC, Server, etc.
"""
__tablename__ = 'computertypes'
computertypeid = db.Column(db.Integer, primary_key=True)
computertype = db.Column(db.String(100), unique=True, nullable=False)
description = db.Column(db.Text)
icon = db.Column(db.String(50), comment='Icon name for UI')
color = db.Column(db.String(20), comment='CSS color for UI/map markers')
def __repr__(self):
return f"<ComputerType {self.computertype}>"
class Computer(BaseModel):
"""
Computer-specific extension data.
Links to core Asset table via assetid.
Stores computer-specific fields like hostname, OS, logged in user, etc.
"""
__tablename__ = 'computers'
computerid = db.Column(db.Integer, primary_key=True)
# Link to core asset
assetid = db.Column(
db.Integer,
db.ForeignKey('assets.assetid', ondelete='CASCADE'),
unique=True,
nullable=False,
index=True
)
# Computer classification
computertypeid = db.Column(
db.Integer,
db.ForeignKey('computertypes.computertypeid'),
nullable=True
)
# Network identity
hostname = db.Column(
db.String(100),
comment='Network hostname'
)
# Operating system
osid = db.Column(
db.Integer,
db.ForeignKey('operatingsystems.osid'),
nullable=True
)
# Hardware make/model (PCs carry vendor + model like machines)
vendorid = db.Column(
db.Integer,
db.ForeignKey('vendors.vendorid'),
nullable=True
)
modelnumberid = db.Column(
db.Integer,
db.ForeignKey('models.modelnumberid'),
nullable=True
)
# Status tracking
loggedinuser = db.Column(db.String(100), nullable=True)
lastreporteddate = db.Column(db.DateTime, nullable=True)
lastboottime = db.Column(db.DateTime, nullable=True)
# Remote access is now modeled per-protocol via the accessmethods
# relationship (AccessProtocol / ComputerAccess), replacing isvnc/iswinrm.
# Relationships
asset = db.relationship(
'Asset',
backref=db.backref('computer', uselist=False, lazy='joined')
)
computertype = db.relationship('ComputerType', backref='computers')
operatingsystem = db.relationship('OperatingSystem', backref='computers')
vendor = db.relationship('Vendor')
model = db.relationship('Model')
# Installed applications (one-to-many)
installedapps = db.relationship(
'ComputerInstalledApp',
back_populates='computer',
cascade='all, delete-orphan',
lazy='dynamic'
)
# Remote-access protocols enabled on this PC (replaces isvnc/iswinrm)
accessmethods = db.relationship(
'ComputerAccess',
back_populates='computer',
cascade='all, delete-orphan',
lazy='selectin'
)
__table_args__ = (
db.Index('idx_computer_type', 'computertypeid'),
db.Index('idx_computer_hostname', 'hostname'),
db.Index('idx_computer_os', 'osid'),
)
def __repr__(self):
return f"<Computer {self.hostname or self.assetid}>"
def to_dict(self):
"""Convert to dictionary with related names."""
result = super().to_dict()
# Add related object names
if self.computertype:
result['computertypename'] = self.computertype.computertype
if self.operatingsystem:
result['osname'] = self.operatingsystem.osname
if self.vendor:
result['vendorname'] = self.vendor.vendor
if self.model:
result['modelname'] = self.model.modelnumber
if self.model.imageurl:
result['imageurl'] = self.model.imageurl
# Names of enabled remote-access protocols (for list badges)
result['accessprotocolnames'] = [
am.protocol.name for am in self.accessmethods
if am.isactive and am.protocol and am.protocol.isactive
]
return result
class ComputerInstalledApp(db.Model):
"""
Junction table for applications installed on computers.
Tracks which applications are installed on which computers,
including version information.
"""
__tablename__ = 'computerinstalledapps'
id = db.Column(db.Integer, primary_key=True)
computerid = db.Column(
db.Integer,
db.ForeignKey('computers.computerid', ondelete='CASCADE'),
nullable=False
)
appid = db.Column(
db.Integer,
db.ForeignKey('applications.appid'),
nullable=False
)
appversionid = db.Column(
db.Integer,
db.ForeignKey('appversions.appversionid'),
nullable=True
)
# Raw version string from automated collection (when no curated AppVersion)
installedversion = db.Column(db.String(100), nullable=True)
isactive = db.Column(db.Boolean, default=True, nullable=False)
installeddate = db.Column(db.DateTime, default=_utcnow)
# Relationships
computer = db.relationship('Computer', back_populates='installedapps')
application = db.relationship('Application')
appversion = db.relationship('AppVersion')
__table_args__ = (
db.UniqueConstraint('computerid', 'appid', name='uq_computer_app'),
db.Index('idx_compapp_computer', 'computerid'),
db.Index('idx_compapp_app', 'appid'),
)
def to_dict(self):
# Curated AppVersion wins; else the raw collected version string.
version = None
if self.appversion is not None:
version = self.appversion.version
elif self.installedversion:
version = self.installedversion
return {
'id': self.id,
'computerid': self.computerid,
'appid': self.appid,
'appname': self.application.appname if self.application else None,
'appdescription': self.application.appdescription if self.application else None,
'appversionid': self.appversionid,
'installedversion': version,
'installeddate': self.installeddate.isoformat() + 'Z' if self.installeddate else None,
'isactive': self.isactive,
}
class AccessProtocol(db.Model):
"""
Catalog of remote-access protocols a PC can expose (VNC, WinRM, RDP, SSH...).
linktemplate builds a connection URL from placeholders {host}, {port},
{scheme}. {host} is the PC hostname joined to the pc_access_domain setting.
Admin-managed; replaces the old fixed isvnc/iswinrm booleans.
"""
__tablename__ = 'accessprotocols'
protocolid = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(50), unique=True, nullable=False)
scheme = db.Column(db.String(20), nullable=False)
defaultport = db.Column(db.Integer, nullable=True)
linktemplate = db.Column(db.String(255), nullable=False)
isactive = db.Column(db.Boolean, default=True, nullable=False)
def to_dict(self):
return {
'protocolid': self.protocolid,
'name': self.name,
'scheme': self.scheme,
'defaultport': self.defaultport,
'linktemplate': self.linktemplate,
'isactive': bool(self.isactive),
}
class ComputerAccess(db.Model):
"""A protocol enabled on a specific PC, with an optional port override."""
__tablename__ = 'computeraccess'
id = db.Column(db.Integer, primary_key=True)
computerid = db.Column(
db.Integer,
db.ForeignKey('computers.computerid', ondelete='CASCADE'),
nullable=False
)
protocolid = db.Column(
db.Integer,
db.ForeignKey('accessprotocols.protocolid'),
nullable=False
)
portoverride = db.Column(db.Integer, nullable=True)
isactive = db.Column(db.Boolean, default=True, nullable=False)
protocol = db.relationship('AccessProtocol')
computer = db.relationship('Computer', back_populates='accessmethods')
__table_args__ = (
db.UniqueConstraint('computerid', 'protocolid', name='uq_computer_protocol'),
db.Index('idx_compaccess_computer', 'computerid'),
)
def to_dict(self):
"""Convert to dictionary."""
return {
'id': self.id,
'computerid': self.computerid,
'appid': self.appid,
'appversionid': self.appversionid,
'isactive': self.isactive,
'installeddate': self.installeddate.isoformat() + 'Z' if self.installeddate else None,
'application': {
'appid': self.application.appid,
'appname': self.application.appname,
'appdescription': self.application.appdescription,
} if self.application else None,
'version': self.appversion.version if self.appversion else None
}
def __repr__(self):
return f"<ComputerInstalledApp computer={self.computerid} app={self.appid}>"