ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Retire the legacy Machine model (ADR-001 cutover)

Browse Source 
The asset/computer model is now the single source of truth. Remove the Machine
instance layer end to end:

- Delete models Machine, MachineStatus, PCType, MachineRelationship,
  InstalledApp, PrinterData; keep MachineType (models.machinetypeid still
  references it).
- Delete the /api/machines, /api/statuses, /api/pctypes blueprints and the
  legacy /api/printers/legacy (PrinterData) blueprint.
- Drop the deprecated communications.machineid column and its FK.
- Migration 7c01 drops tables machines, machinestatuses, pctypes,
  machinerelationships, installedapps, printerdata (idempotent).
- Fix remaining readers (applications install counts) to ComputerInstalledApp.
- Frontend: remove dead machinesApi/statusesApi/pctypesApi wrappers; repoint
  the PC Types settings page at computer types.

143 tests pass; all asset/computer/dashboard/report/collector endpoints 200.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
cproudlock
2026-06-26 09:59:17 -04:00
parent cb47b9087e
commit b0a0670fcd
23 changed files with 344 additions and 2047 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
shopdb/core/api/__init__.py
View File

@@ -2,10 +2,7 @@
from .auth import auth_bp
from .assets import assets_bp
from .machines import machines_bp
from .machinetypes import machinetypes_bp
from .pctypes import pctypes_bp
from .statuses import statuses_bp
from .vendors import vendors_bp
from .models import models_bp
from .businessunits import businessunits_bp
@@ -26,10 +23,7 @@ from .users import users_bp
__all__ = [
'auth_bp',
'assets_bp',
'machines_bp',
'machinetypes_bp',
'pctypes_bp',
'statuses_bp',
'vendors_bp',
'models_bp',
'businessunits_bp',

6
shopdb/core/api/applications.py
View File

@@ -66,7 +66,8 @@ def list_applications():
}
else:
app_dict['supportteam'] = None
app_dict['installedcount'] = app.installed_on.filter_by(isactive=True).count()
app_dict['installedcount'] = ComputerInstalledApp.query.filter_by(
appid=app.appid, isactive=True).count()
data.append(app_dict)
return paginated_response(data, page, per_page, total)
@@ -96,7 +97,8 @@ def get_application(app_id: int):
else:
data['supportteam'] = None
data['versions'] = [v.to_dict() for v in app.versions.filter_by(isactive=True).order_by(AppVersion.version.desc()).all()]
data['installedcount'] = app.installed_on.filter_by(isactive=True).count()
data['installedcount'] = ComputerInstalledApp.query.filter_by(
appid=app.appid, isactive=True).count()
return success_response(data)

641
shopdb/core/api/machines.py
View File

@@ -1,641 +0,0 @@
"""
Machines API endpoints.
DEPRECATED: This API is deprecated and will be removed in a future version.
Please migrate to the new asset-based APIs:
- /api/assets - Unified asset queries
- /api/equipment - Equipment CRUD
- /api/computers - Computers CRUD
- /api/network - Network devices CRUD
- /api/printers - Printers CRUD
"""
import logging
from functools import wraps
from flask import Blueprint, request, g
from flask_jwt_extended import jwt_required, current_user
from shopdb.extensions import db
from shopdb.core.models import Machine, MachineType, AuditLog
from shopdb.core.models.relationship import MachineRelationship, RelationshipType
from shopdb.utils.responses import (
success_response,
error_response,
paginated_response,
ErrorCodes
)
from shopdb.utils.pagination import get_pagination_params, paginate_query
logger = logging.getLogger(__name__)
machines_bp = Blueprint('machines', __name__)
def add_deprecation_headers(f):
"""Decorator to add deprecation headers to responses."""
@wraps(f)
def decorated_function(*args, **kwargs):
response = f(*args, **kwargs)
# Add deprecation headers
if hasattr(response, 'headers'):
response.headers['X-Deprecated'] = 'true'
response.headers['X-Deprecated-Message'] = (
'This endpoint is deprecated. '
'Please migrate to /api/assets, /api/equipment, /api/computers, /api/network, or /api/printers.'
)
response.headers['Sunset'] = '2026-12-31' # Target sunset date
# Log deprecation warning (once per request)
if not getattr(g, '_deprecation_logged', False):
logger.warning(
f"Deprecated /api/machines endpoint called: {request.method} {request.path}"
)
g._deprecation_logged = True
return response
return decorated_function
@machines_bp.route('', methods=['GET'])
@jwt_required(optional=True)
@add_deprecation_headers
def list_machines():
"""
List all machines with filtering and pagination.
Query params:
page: int (default 1)
per_page: int (default 20, max 100)
machinetype: int (filter by type ID)
pctype: int (filter by PC type ID)
businessunit: int (filter by business unit ID)
status: int (filter by status ID)
category: str (Equipment, PC, Network)
search: str (search in machinenumber, alias, hostname)
active: bool (default true)
sort: str (field name, prefix with - for desc)
"""
page, per_page = get_pagination_params(request)
# Build query
query = Machine.query
# Apply filters
if request.args.get('active', 'true').lower() != 'false':
query = query.filter(Machine.isactive == True)
if machinetype_id := request.args.get('machinetype', type=int):
query = query.filter(Machine.machinetypeid == machinetype_id)
if pctype_id := request.args.get('pctype', type=int):
query = query.filter(Machine.pctypeid == pctype_id)
if businessunit_id := request.args.get('businessunit', type=int):
query = query.filter(Machine.businessunitid == businessunit_id)
if status_id := request.args.get('status', type=int):
query = query.filter(Machine.statusid == status_id)
if category := request.args.get('category'):
query = query.join(MachineType).filter(MachineType.category == category)
if search := request.args.get('search'):
search_term = f'%{search}%'
query = query.filter(
db.or_(
Machine.machinenumber.ilike(search_term),
Machine.alias.ilike(search_term),
Machine.hostname.ilike(search_term),
Machine.serialnumber.ilike(search_term)
)
)
# Filter for machines with map positions
if request.args.get('hasmap', '').lower() == 'true':
query = query.filter(
Machine.mapleft.isnot(None),
Machine.maptop.isnot(None)
)
# Apply sorting
sort_field = request.args.get('sort', 'machinenumber')
desc = sort_field.startswith('-')
if desc:
sort_field = sort_field[1:]
if hasattr(Machine, sort_field):
order = getattr(Machine, sort_field)
query = query.order_by(order.desc() if desc else order)
# For map view, allow fetching all machines without pagination limit
include_map_extras = request.args.get('hasmap', '').lower() == 'true'
fetch_all = request.args.get('all', '').lower() == 'true'
if include_map_extras and fetch_all:
# Get all map machines without pagination
items = query.all()
total = len(items)
else:
# Normal pagination
items, total = paginate_query(query, page, per_page)
# Convert to dicts with relationships
data = []
for m in items:
d = m.to_dict()
# Get machinetype from model (single source of truth)
mt = m.derived_machinetype
d['machinetype'] = mt.machinetype if mt else None
d['machinetypeid'] = mt.machinetypeid if mt else None
d['category'] = mt.category if mt else None
d['status'] = m.status.status if m.status else None
d['statusid'] = m.statusid
d['businessunit'] = m.businessunit.businessunit if m.businessunit else None
d['businessunitid'] = m.businessunitid
d['vendor'] = m.vendor.vendor if m.vendor else None
d['model'] = m.model.modelnumber if m.model else None
d['pctype'] = m.pctype.pctype if m.pctype else None
d['serialnumber'] = m.serialnumber
d['isvnc'] = m.isvnc
d['iswinrm'] = m.iswinrm
# Include extra fields for map view
if include_map_extras:
# Get primary IP address from communications
primary_comm = next(
(c for c in m.communications if c.isprimary and c.ipaddress),
None
)
if not primary_comm:
# Fall back to first communication with IP
primary_comm = next(
(c for c in m.communications if c.ipaddress),
None
)
d['ipaddress'] = primary_comm.ipaddress if primary_comm else None
# Get connected PC (parent machine that is a PC)
connected_pc = None
for rel in m.parent_relationships:
if rel.parent_machine and rel.parent_machine.is_pc:
connected_pc = rel.parent_machine.machinenumber
break
d['connected_pc'] = connected_pc
data.append(d)
return paginated_response(data, page, per_page, total)
@machines_bp.route('/<int:machine_id>', methods=['GET'])
@jwt_required(optional=True)
@add_deprecation_headers
def get_machine(machine_id: int):
"""Get a single machine by ID."""
machine = Machine.query.get(machine_id)
if not machine:
return error_response(
ErrorCodes.NOT_FOUND,
f'Machine with ID {machine_id} not found',
http_code=404
)
data = machine.to_dict()
# Add related data - machinetype comes from model (single source of truth)
mt = machine.derived_machinetype
data['machinetype'] = mt.to_dict() if mt else None
data['pctype'] = machine.pctype.to_dict() if machine.pctype else None
data['status'] = machine.status.to_dict() if machine.status else None
data['businessunit'] = machine.businessunit.to_dict() if machine.businessunit else None
data['vendor'] = machine.vendor.to_dict() if machine.vendor else None
data['model'] = machine.model.to_dict() if machine.model else None
data['location'] = machine.location.to_dict() if machine.location else None
data['operatingsystem'] = machine.operatingsystem.to_dict() if machine.operatingsystem else None
# Add communications
data['communications'] = [c.to_dict() for c in machine.communications.all()]
return success_response(data)
@machines_bp.route('', methods=['POST'])
@jwt_required()
@add_deprecation_headers
def create_machine():
"""Create a new machine."""
data = request.get_json()
if not data:
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
if not data.get('machinenumber'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'machinenumber is required')
if not data.get('modelnumberid'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'modelnumberid is required (determines machine type)')
# Check for duplicate machinenumber
if Machine.query.filter_by(machinenumber=data['machinenumber']).first():
return error_response(
ErrorCodes.CONFLICT,
f"Machine number '{data['machinenumber']}' already exists",
http_code=409
)
# Create machine
allowed_fields = [
'machinenumber', 'alias', 'hostname', 'serialnumber',
'machinetypeid', 'pctypeid', 'businessunitid', 'modelnumberid',
'vendorid', 'statusid', 'locationid', 'osid',
'mapleft', 'maptop', 'islocationonly',
'loggedinuser', 'isvnc', 'iswinrm', 'isshopfloor',
'requiresmanualconfig', 'notes'
]
machine_data = {k: v for k, v in data.items() if k in allowed_fields}
machine = Machine(**machine_data)
machine.createdby = current_user.username
db.session.add(machine)
db.session.flush()
# Audit log
AuditLog.log('created', 'Machine', entityid=machine.machineid,
entityname=machine.machinenumber)
db.session.commit()
return success_response(
machine.to_dict(),
message='Machine created successfully',
http_code=201
)
@machines_bp.route('/<int:machine_id>', methods=['PUT'])
@jwt_required()
@add_deprecation_headers
def update_machine(machine_id: int):
"""Update an existing machine."""
machine = Machine.query.get(machine_id)
if not machine:
return error_response(
ErrorCodes.NOT_FOUND,
f'Machine with ID {machine_id} not found',
http_code=404
)
data = request.get_json()
if not data:
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
# Check for duplicate machinenumber if changed
if 'machinenumber' in data and data['machinenumber'] != machine.machinenumber:
existing = Machine.query.filter_by(machinenumber=data['machinenumber']).first()
if existing:
return error_response(
ErrorCodes.CONFLICT,
f"Machine number '{data['machinenumber']}' already exists",
http_code=409
)
# Update allowed fields
allowed_fields = [
'machinenumber', 'alias', 'hostname', 'serialnumber',
'machinetypeid', 'pctypeid', 'businessunitid', 'modelnumberid',
'vendorid', 'statusid', 'locationid', 'osid',
'mapleft', 'maptop', 'islocationonly',
'loggedinuser', 'isvnc', 'iswinrm', 'isshopfloor',
'requiresmanualconfig', 'notes', 'isactive'
]
# Track changes for audit log
changes = {}
for key, value in data.items():
if key in allowed_fields:
old_val = getattr(machine, key)
if old_val != value:
changes[key] = {'old': old_val, 'new': value}
setattr(machine, key, value)
machine.modifiedby = current_user.username
# Audit log if there were changes
if changes:
AuditLog.log('updated', 'Machine', entityid=machine.machineid,
entityname=machine.machinenumber, changes=changes)
db.session.commit()
return success_response(machine.to_dict(), message='Machine updated successfully')
@machines_bp.route('/<int:machine_id>', methods=['DELETE'])
@jwt_required()
@add_deprecation_headers
def delete_machine(machine_id: int):
"""Soft delete a machine."""
machine = Machine.query.get(machine_id)
if not machine:
return error_response(
ErrorCodes.NOT_FOUND,
f'Machine with ID {machine_id} not found',
http_code=404
)
machine.soft_delete(deleted_by=current_user.username)
# Audit log
AuditLog.log('deleted', 'Machine', entityid=machine.machineid,
entityname=machine.machinenumber)
db.session.commit()
return success_response(message='Machine deleted successfully')
@machines_bp.route('/<int:machine_id>/communications', methods=['GET'])
@jwt_required(optional=True)
@add_deprecation_headers
def get_machine_communications(machine_id: int):
"""Get all communications for a machine."""
machine = Machine.query.get(machine_id)
if not machine:
return error_response(
ErrorCodes.NOT_FOUND,
f'Machine with ID {machine_id} not found',
http_code=404
)
comms = [c.to_dict() for c in machine.communications.all()]
return success_response(comms)
@machines_bp.route('/<int:machine_id>/communication', methods=['PUT'])
@jwt_required()
@add_deprecation_headers
def update_machine_communication(machine_id: int):
"""Update machine communication (IP address)."""
from shopdb.core.models.communication import Communication, CommunicationType
machine = Machine.query.get(machine_id)
if not machine:
return error_response(
ErrorCodes.NOT_FOUND,
f'Machine with ID {machine_id} not found',
http_code=404
)
data = request.get_json()
if not data:
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
# Get or create IP communication type
ip_comtype = CommunicationType.query.filter_by(comtype='IP').first()
if not ip_comtype:
ip_comtype = CommunicationType(comtype='IP', description='IP Network')
db.session.add(ip_comtype)
db.session.flush()
# Find existing primary communication or create new one
comms = list(machine.communications.all())
comm = next((c for c in comms if c.isprimary), None)
if not comm:
comm = next((c for c in comms if c.comtypeid == ip_comtype.comtypeid), None)
if not comm:
comm = Communication(machineid=machine_id, comtypeid=ip_comtype.comtypeid)
db.session.add(comm)
# Update fields
if 'ipaddress' in data:
comm.ipaddress = data['ipaddress']
if 'isprimary' in data:
comm.isprimary = data['isprimary']
if 'macaddress' in data:
comm.macaddress = data['macaddress']
db.session.commit()
return success_response({
'communicationid': comm.communicationid,
'ipaddress': comm.ipaddress,
'isprimary': comm.isprimary,
}, message='Communication updated')
# ==================== Machine Relationships ====================
@machines_bp.route('/<int:machine_id>/relationships', methods=['GET'])
@jwt_required(optional=True)
@add_deprecation_headers
def get_machine_relationships(machine_id: int):
"""Get all relationships for a machine (both parent and child)."""
machine = Machine.query.get(machine_id)
if not machine:
return error_response(
ErrorCodes.NOT_FOUND,
f'Machine with ID {machine_id} not found',
http_code=404
)
relationships = []
my_category = machine.machinetype.category if machine.machinetype else None
seen_ids = set()
# Get all relationships involving this machine
all_rels = list(machine.child_relationships) + list(machine.parent_relationships)
for rel in all_rels:
if rel.relationshipid in seen_ids:
continue
seen_ids.add(rel.relationshipid)
# Determine the related machine (the one that isn't us)
if rel.parentmachineid == machine.machineid:
related = rel.child_machine
else:
related = rel.parent_machine
related_category = related.machinetype.category if related and related.machinetype else None
rel_type = rel.relationship_type.relationshiptype if rel.relationship_type else None
# Determine direction based on relationship type and categories
if rel_type == 'Controls':
# PC controls Equipment - determine from categories
if my_category == 'PC':
direction = 'controls'
else:
direction = 'controlled_by'
elif rel_type == 'Dualpath':
direction = 'dualpath_partner'
else:
# For other types, use parent/child
if rel.parentmachineid == machine.machineid:
direction = 'controls'
else:
direction = 'controlled_by'
relationships.append({
'relationshipid': rel.relationshipid,
'direction': direction,
'relatedmachineid': related.machineid if related else None,
'relatedmachinenumber': related.machinenumber if related else None,
'relatedmachinealias': related.alias if related else None,
'relatedcategory': related_category,
'relationshiptype': rel_type,
'relationshiptypeid': rel.relationshiptypeid,
'notes': rel.notes
})
return success_response(relationships)
@machines_bp.route('/<int:machine_id>/relationships', methods=['POST'])
@jwt_required()
@add_deprecation_headers
def create_machine_relationship(machine_id: int):
"""Create a relationship for a machine."""
machine = Machine.query.get(machine_id)
if not machine:
return error_response(
ErrorCodes.NOT_FOUND,
f'Machine with ID {machine_id} not found',
http_code=404
)
data = request.get_json()
if not data:
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
related_machine_id = data.get('relatedmachineid')
relationship_type_id = data.get('relationshiptypeid')
direction = data.get('direction', 'controlled_by') # 'controls' or 'controlled_by'
if not related_machine_id:
return error_response(ErrorCodes.VALIDATION_ERROR, 'relatedmachineid is required')
if not relationship_type_id:
return error_response(ErrorCodes.VALIDATION_ERROR, 'relationshiptypeid is required')
related_machine = Machine.query.get(related_machine_id)
if not related_machine:
return error_response(
ErrorCodes.NOT_FOUND,
f'Related machine with ID {related_machine_id} not found',
http_code=404
)
# Determine parent/child based on direction
if direction == 'controls':
parent_id = machine_id
child_id = related_machine_id
else: # controlled_by
parent_id = related_machine_id
child_id = machine_id
# Check if relationship already exists
existing = MachineRelationship.query.filter_by(
parentmachineid=parent_id,
childmachineid=child_id,
relationshiptypeid=relationship_type_id
).first()
if existing:
return error_response(
ErrorCodes.CONFLICT,
'This relationship already exists',
http_code=409
)
relationship = MachineRelationship(
parentmachineid=parent_id,
childmachineid=child_id,
relationshiptypeid=relationship_type_id,
notes=data.get('notes')
)
db.session.add(relationship)
db.session.commit()
return success_response({
'relationshipid': relationship.relationshipid,
'parentmachineid': relationship.parentmachineid,
'childmachineid': relationship.childmachineid,
'relationshiptypeid': relationship.relationshiptypeid
}, message='Relationship created successfully', http_code=201)
@machines_bp.route('/relationships/<int:relationship_id>', methods=['DELETE'])
@jwt_required()
@add_deprecation_headers
def delete_machine_relationship(relationship_id: int):
"""Delete a machine relationship."""
relationship = MachineRelationship.query.get(relationship_id)
if not relationship:
return error_response(
ErrorCodes.NOT_FOUND,
f'Relationship with ID {relationship_id} not found',
http_code=404
)
db.session.delete(relationship)
db.session.commit()
return success_response(message='Relationship deleted successfully')
@machines_bp.route('/relationshiptypes', methods=['GET'])
@jwt_required(optional=True)
@add_deprecation_headers
def list_relationship_types():
"""List all relationship types."""
types = RelationshipType.query.order_by(RelationshipType.relationshiptype).all()
return success_response([{
'relationshiptypeid': t.relationshiptypeid,
'relationshiptype': t.relationshiptype,
'description': t.description
} for t in types])
@machines_bp.route('/relationshiptypes', methods=['POST'])
@jwt_required()
@add_deprecation_headers
def create_relationship_type():
"""Create a new relationship type."""
data = request.get_json()
if not data:
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
if not data.get('relationshiptype'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'relationshiptype is required')
existing = RelationshipType.query.filter_by(relationshiptype=data['relationshiptype']).first()
if existing:
return error_response(
ErrorCodes.CONFLICT,
f"Relationship type '{data['relationshiptype']}' already exists",
http_code=409
)
rel_type = RelationshipType(
relationshiptype=data['relationshiptype'],
description=data.get('description')
)
db.session.add(rel_type)
db.session.commit()
return success_response({
'relationshiptypeid': rel_type.relationshiptypeid,
'relationshiptype': rel_type.relationshiptype,
'description': rel_type.description
}, message='Relationship type created successfully', http_code=201)

8
shopdb/core/api/machinetypes.py
View File

@@ -133,12 +133,12 @@ def delete_machinetype(type_id: int):
http_code=404
)
# Check if any machines use this type
from shopdb.core.models import Machine
if Machine.query.filter_by(machinetypeid=type_id, isactive=True).first():
# Check if any model uses this type
from shopdb.core.models import Model
if Model.query.filter_by(machinetypeid=type_id).first():
return error_response(
ErrorCodes.CONFLICT,
'Cannot delete machine type: machines are using it',
'Cannot delete machine type: models are using it',
http_code=409
)

141
shopdb/core/api/pctypes.py
View File

@@ -1,141 +0,0 @@
"""PC Types API endpoints - Full CRUD."""
from flask import Blueprint, request
from flask_jwt_extended import jwt_required
from shopdb.extensions import db
from shopdb.core.models import PCType
from shopdb.utils.responses import (
success_response,
error_response,
paginated_response,
ErrorCodes
)
from shopdb.utils.pagination import get_pagination_params, paginate_query
pctypes_bp = Blueprint('pctypes', __name__)
@pctypes_bp.route('', methods=['GET'])
@jwt_required(optional=True)
def list_pctypes():
"""List all PC types."""
page, per_page = get_pagination_params(request)
query = PCType.query
if request.args.get('active', 'true').lower() != 'false':
query = query.filter(PCType.isactive == True)
if search := request.args.get('search'):
query = query.filter(PCType.pctype.ilike(f'%{search}%'))
query = query.order_by(PCType.pctype)
items, total = paginate_query(query, page, per_page)
data = [pt.to_dict() for pt in items]
return paginated_response(data, page, per_page, total)
@pctypes_bp.route('/<int:type_id>', methods=['GET'])
@jwt_required(optional=True)
def get_pctype(type_id: int):
"""Get a single PC type."""
pt = PCType.query.get(type_id)
if not pt:
return error_response(
ErrorCodes.NOT_FOUND,
f'PC type with ID {type_id} not found',
http_code=404
)
return success_response(pt.to_dict())
@pctypes_bp.route('', methods=['POST'])
@jwt_required()
def create_pctype():
"""Create a new PC type."""
data = request.get_json()
if not data or not data.get('pctype'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'pctype is required')
if PCType.query.filter_by(pctype=data['pctype']).first():
return error_response(
ErrorCodes.CONFLICT,
f"PC type '{data['pctype']}' already exists",
http_code=409
)
pt = PCType(
pctype=data['pctype'],
description=data.get('description')
)
db.session.add(pt)
db.session.commit()
return success_response(pt.to_dict(), message='PC type created', http_code=201)
@pctypes_bp.route('/<int:type_id>', methods=['PUT'])
@jwt_required()
def update_pctype(type_id: int):
"""Update a PC type."""
pt = PCType.query.get(type_id)
if not pt:
return error_response(
ErrorCodes.NOT_FOUND,
f'PC 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 'pctype' in data and data['pctype'] != pt.pctype:
if PCType.query.filter_by(pctype=data['pctype']).first():
return error_response(
ErrorCodes.CONFLICT,
f"PC type '{data['pctype']}' already exists",
http_code=409
)
for key in ['pctype', 'description', 'isactive']:
if key in data:
setattr(pt, key, data[key])
db.session.commit()
return success_response(pt.to_dict(), message='PC type updated')
@pctypes_bp.route('/<int:type_id>', methods=['DELETE'])
@jwt_required()
def delete_pctype(type_id: int):
"""Delete (deactivate) a PC type."""
pt = PCType.query.get(type_id)
if not pt:
return error_response(
ErrorCodes.NOT_FOUND,
f'PC type with ID {type_id} not found',
http_code=404
)
from shopdb.core.models import Machine
if Machine.query.filter_by(pctypeid=type_id, isactive=True).first():
return error_response(
ErrorCodes.CONFLICT,
'Cannot delete PC type: machines are using it',
http_code=409
)
pt.isactive = False
db.session.commit()
return success_response(message='PC type deleted')

139
shopdb/core/api/statuses.py
View File

@@ -1,139 +0,0 @@
"""Machine Statuses API endpoints - Full CRUD."""
from flask import Blueprint, request
from flask_jwt_extended import jwt_required
from shopdb.extensions import db
from shopdb.core.models import MachineStatus
from shopdb.utils.responses import (
success_response,
error_response,
paginated_response,
ErrorCodes
)
from shopdb.utils.pagination import get_pagination_params, paginate_query
statuses_bp = Blueprint('statuses', __name__)
@statuses_bp.route('', methods=['GET'])
@jwt_required(optional=True)
def list_statuses():
"""List all machine statuses."""
page, per_page = get_pagination_params(request)
query = MachineStatus.query
if request.args.get('active', 'true').lower() != 'false':
query = query.filter(MachineStatus.isactive == True)
query = query.order_by(MachineStatus.status)
items, total = paginate_query(query, page, per_page)
data = [s.to_dict() for s in items]
return paginated_response(data, page, per_page, total)
@statuses_bp.route('/<int:status_id>', methods=['GET'])
@jwt_required(optional=True)
def get_status(status_id: int):
"""Get a single status."""
s = MachineStatus.query.get(status_id)
if not s:
return error_response(
ErrorCodes.NOT_FOUND,
f'Status with ID {status_id} not found',
http_code=404
)
return success_response(s.to_dict())
@statuses_bp.route('', methods=['POST'])
@jwt_required()
def create_status():
"""Create a new status."""
data = request.get_json()
if not data or not data.get('status'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'status is required')
if MachineStatus.query.filter_by(status=data['status']).first():
return error_response(
ErrorCodes.CONFLICT,
f"Status '{data['status']}' already exists",
http_code=409
)
s = MachineStatus(
status=data['status'],
description=data.get('description'),
color=data.get('color')
)
db.session.add(s)
db.session.commit()
return success_response(s.to_dict(), message='Status created', http_code=201)
@statuses_bp.route('/<int:status_id>', methods=['PUT'])
@jwt_required()
def update_status(status_id: int):
"""Update a status."""
s = MachineStatus.query.get(status_id)
if not s:
return error_response(
ErrorCodes.NOT_FOUND,
f'Status with ID {status_id} not found',
http_code=404
)
data = request.get_json()
if not data:
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
if 'status' in data and data['status'] != s.status:
if MachineStatus.query.filter_by(status=data['status']).first():
return error_response(
ErrorCodes.CONFLICT,
f"Status '{data['status']}' already exists",
http_code=409
)
for key in ['status', 'description', 'color', 'isactive']:
if key in data:
setattr(s, key, data[key])
db.session.commit()
return success_response(s.to_dict(), message='Status updated')
@statuses_bp.route('/<int:status_id>', methods=['DELETE'])
@jwt_required()
def delete_status(status_id: int):
"""Delete (deactivate) a status."""
s = MachineStatus.query.get(status_id)
if not s:
return error_response(
ErrorCodes.NOT_FOUND,
f'Status with ID {status_id} not found',
http_code=404
)
from shopdb.core.models import Machine
if Machine.query.filter_by(statusid=status_id, isactive=True).first():
return error_response(
ErrorCodes.CONFLICT,
'Cannot delete status: machines are using it',
http_code=409
)
s.isactive = False
db.session.commit()
return success_response(message='Status deleted')