Initial commit: Shop Database Flask Application

Flask backend with Vue 3 frontend for shop floor machine management.
Includes database schema export for MySQL shopdb_flask database.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
cproudlock
2026-01-13 16:07:34 -05:00
commit 1196de6e88
188 changed files with 19921 additions and 0 deletions

189
shopdb/__init__.py Normal file
View File

@@ -0,0 +1,189 @@
"""Flask application factory."""
import os
import logging
from flask import Flask, send_from_directory
from .config import config
from .extensions import db, migrate, jwt, cors, ma, init_extensions
from .plugins import plugin_manager
def create_app(config_name: str = None) -> Flask:
"""
Application factory.
Args:
config_name: Configuration name ('development', 'production', 'testing')
Returns:
Configured Flask application
"""
if config_name is None:
config_name = os.environ.get('FLASK_ENV', 'development')
app = Flask(__name__, instance_relative_config=True)
# Load configuration
app.config.from_object(config.get(config_name, config['default']))
# Load instance config if exists
app.config.from_pyfile('config.py', silent=True)
# Ensure instance folder exists
os.makedirs(app.instance_path, exist_ok=True)
# Configure logging
configure_logging(app)
# Initialize extensions
init_extensions(app)
# Initialize plugin manager
with app.app_context():
plugin_manager.init_app(app, db)
# Register core blueprints
register_blueprints(app)
# Register CLI commands
register_cli_commands(app)
# Register error handlers
register_error_handlers(app)
# Serve Vue frontend
register_frontend_routes(app)
# JWT user loader (identity is a string in JWT, convert to int for DB lookup)
@jwt.user_lookup_loader
def user_lookup_callback(_jwt_header, jwt_data):
from .core.models import User
identity = jwt_data["sub"]
return User.query.get(int(identity))
return app
def register_blueprints(app: Flask):
"""Register core API blueprints."""
from .core.api import (
auth_bp,
machines_bp,
machinetypes_bp,
pctypes_bp,
statuses_bp,
vendors_bp,
models_bp,
businessunits_bp,
locations_bp,
operatingsystems_bp,
dashboard_bp,
applications_bp,
knowledgebase_bp,
search_bp,
)
api_prefix = '/api'
app.register_blueprint(auth_bp, url_prefix=f'{api_prefix}/auth')
app.register_blueprint(machines_bp, url_prefix=f'{api_prefix}/machines')
app.register_blueprint(machinetypes_bp, url_prefix=f'{api_prefix}/machinetypes')
app.register_blueprint(pctypes_bp, url_prefix=f'{api_prefix}/pctypes')
app.register_blueprint(statuses_bp, url_prefix=f'{api_prefix}/statuses')
app.register_blueprint(vendors_bp, url_prefix=f'{api_prefix}/vendors')
app.register_blueprint(models_bp, url_prefix=f'{api_prefix}/models')
app.register_blueprint(businessunits_bp, url_prefix=f'{api_prefix}/businessunits')
app.register_blueprint(locations_bp, url_prefix=f'{api_prefix}/locations')
app.register_blueprint(operatingsystems_bp, url_prefix=f'{api_prefix}/operatingsystems')
app.register_blueprint(dashboard_bp, url_prefix=f'{api_prefix}/dashboard')
app.register_blueprint(applications_bp, url_prefix=f'{api_prefix}/applications')
app.register_blueprint(knowledgebase_bp, url_prefix=f'{api_prefix}/knowledgebase')
app.register_blueprint(search_bp, url_prefix=f'{api_prefix}/search')
def register_cli_commands(app: Flask):
"""Register Flask CLI commands."""
from .plugins.cli import plugin_cli
from .cli import db_cli, seed_cli
app.cli.add_command(plugin_cli)
app.cli.add_command(db_cli)
app.cli.add_command(seed_cli)
def register_error_handlers(app: Flask):
"""Register error handlers."""
from .utils.responses import error_response, ErrorCodes
from .exceptions import ShopDBException
@app.errorhandler(ShopDBException)
def handle_shopdb_exception(error):
http_codes = {
'NOT_FOUND': 404,
'UNAUTHORIZED': 401,
'FORBIDDEN': 403,
'CONFLICT': 409,
'VALIDATION_ERROR': 400,
}
http_code = http_codes.get(error.code, 400)
return error_response(
error.code,
error.message,
details=error.details,
http_code=http_code
)
@app.errorhandler(404)
def not_found_error(error):
return error_response(
ErrorCodes.NOT_FOUND,
'Resource not found',
http_code=404
)
@app.errorhandler(500)
def internal_error(error):
return error_response(
ErrorCodes.INTERNAL_ERROR,
'An internal error occurred',
http_code=500
)
@app.errorhandler(401)
def unauthorized_error(error):
return error_response(
ErrorCodes.UNAUTHORIZED,
'Authentication required',
http_code=401
)
def register_frontend_routes(app: Flask):
"""Serve Vue frontend static files."""
frontend_dist = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'frontend', 'dist')
@app.route('/', defaults={'path': ''})
@app.route('/<path:path>')
def serve_frontend(path):
# Don't serve API routes as frontend
if path.startswith('api/'):
from .utils.responses import error_response, ErrorCodes
return error_response(ErrorCodes.NOT_FOUND, 'API endpoint not found', http_code=404)
# Serve static assets
if path and os.path.exists(os.path.join(frontend_dist, path)):
return send_from_directory(frontend_dist, path)
# Serve index.html for SPA routing
return send_from_directory(frontend_dist, 'index.html')
def configure_logging(app: Flask):
"""Configure application logging."""
log_level = app.config.get('LOG_LEVEL', 'INFO')
logging.basicConfig(
level=getattr(logging, log_level),
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

147
shopdb/cli/__init__.py Normal file
View File

@@ -0,0 +1,147 @@
"""Flask CLI commands."""
import click
from flask.cli import with_appcontext
@click.group('db-utils')
def db_cli():
"""Database utility commands."""
pass
@db_cli.command('create-all')
@with_appcontext
def create_all():
"""Create all database tables."""
from shopdb.extensions import db
db.create_all()
click.echo(click.style("All tables created.", fg='green'))
@db_cli.command('drop-all')
@click.confirmation_option(prompt='This will delete ALL data. Are you sure?')
@with_appcontext
def drop_all():
"""Drop all database tables."""
from shopdb.extensions import db
db.drop_all()
click.echo(click.style("All tables dropped.", fg='yellow'))
@click.group('seed')
def seed_cli():
"""Database seeding commands."""
pass
@seed_cli.command('reference-data')
@with_appcontext
def seed_reference_data():
"""Seed reference data (machine types, statuses, etc.)."""
from shopdb.extensions import db
from shopdb.core.models import MachineType, MachineStatus, OperatingSystem
from shopdb.core.models.relationship import RelationshipType
# Machine types
machine_types = [
{'machinetype': 'CNC Mill', 'category': 'Equipment', 'description': 'CNC Milling Machine'},
{'machinetype': 'CNC Lathe', 'category': 'Equipment', 'description': 'CNC Lathe'},
{'machinetype': 'CMM', 'category': 'Equipment', 'description': 'Coordinate Measuring Machine'},
{'machinetype': 'EDM', 'category': 'Equipment', 'description': 'Electrical Discharge Machine'},
{'machinetype': 'Grinder', 'category': 'Equipment', 'description': 'Grinding Machine'},
{'machinetype': 'Inspection Station', 'category': 'Equipment', 'description': 'Inspection Station'},
{'machinetype': 'Desktop PC', 'category': 'PC', 'description': 'Desktop Computer'},
{'machinetype': 'Laptop', 'category': 'PC', 'description': 'Laptop Computer'},
{'machinetype': 'Shopfloor PC', 'category': 'PC', 'description': 'Shopfloor Computer'},
{'machinetype': 'Server', 'category': 'Network', 'description': 'Server'},
{'machinetype': 'Switch', 'category': 'Network', 'description': 'Network Switch'},
{'machinetype': 'Access Point', 'category': 'Network', 'description': 'Wireless Access Point'},
]
for mt_data in machine_types:
existing = MachineType.query.filter_by(machinetype=mt_data['machinetype']).first()
if not existing:
mt = MachineType(**mt_data)
db.session.add(mt)
# Machine statuses
statuses = [
{'status': 'In Use', 'description': 'Currently in use', 'color': '#28a745'},
{'status': 'Spare', 'description': 'Available as spare', 'color': '#17a2b8'},
{'status': 'Retired', 'description': 'No longer in use', 'color': '#6c757d'},
{'status': 'In Repair', 'description': 'Currently being repaired', 'color': '#ffc107'},
{'status': 'Pending', 'description': 'Pending installation', 'color': '#007bff'},
]
for s_data in statuses:
existing = MachineStatus.query.filter_by(status=s_data['status']).first()
if not existing:
s = MachineStatus(**s_data)
db.session.add(s)
# Operating systems
os_list = [
{'osname': 'Windows 10', 'osversion': '10.0'},
{'osname': 'Windows 11', 'osversion': '11.0'},
{'osname': 'Windows Server 2019', 'osversion': '2019'},
{'osname': 'Windows Server 2022', 'osversion': '2022'},
{'osname': 'Linux', 'osversion': 'Various'},
]
for os_data in os_list:
existing = OperatingSystem.query.filter_by(osname=os_data['osname']).first()
if not existing:
os_obj = OperatingSystem(**os_data)
db.session.add(os_obj)
# Connection types (how PC connects to equipment)
connection_types = [
{'relationshiptype': 'Serial Cable', 'description': 'RS-232 or similar serial connection'},
{'relationshiptype': 'Direct Ethernet', 'description': 'Direct network cable (airgapped)'},
{'relationshiptype': 'USB', 'description': 'USB connection'},
{'relationshiptype': 'WiFi', 'description': 'Wireless network connection'},
{'relationshiptype': 'Dualpath', 'description': 'Redundant/failover network path'},
]
for ct_data in connection_types:
existing = RelationshipType.query.filter_by(relationshiptype=ct_data['relationshiptype']).first()
if not existing:
ct = RelationshipType(**ct_data)
db.session.add(ct)
db.session.commit()
click.echo(click.style("Reference data seeded.", fg='green'))
@seed_cli.command('test-user')
@with_appcontext
def seed_test_user():
"""Create a test admin user."""
from shopdb.extensions import db
from shopdb.core.models import User, Role
from werkzeug.security import generate_password_hash
# Create admin role if not exists
admin_role = Role.query.filter_by(rolename='admin').first()
if not admin_role:
admin_role = Role(rolename='admin', description='Administrator')
db.session.add(admin_role)
# Create test user
test_user = User.query.filter_by(username='admin').first()
if not test_user:
test_user = User(
username='admin',
email='admin@localhost',
passwordhash=generate_password_hash('admin123'),
isactive=True
)
test_user.roles.append(admin_role)
db.session.add(test_user)
db.session.commit()
click.echo(click.style("Test user created: admin / admin123", fg='green'))
else:
click.echo(click.style("Test user already exists", fg='yellow'))

82
shopdb/config.py Normal file
View File

@@ -0,0 +1,82 @@
"""Flask application configuration."""
import os
from datetime import timedelta
class Config:
"""Base configuration."""
# Flask
SECRET_KEY = os.environ.get('SECRET_KEY', 'dev-secret-key-change-in-production')
# SQLAlchemy
SQLALCHEMY_DATABASE_URI = os.environ.get(
'DATABASE_URL',
'mysql+pymysql://root:password@localhost:3306/shopdb_flask'
)
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ENGINE_OPTIONS = {
'pool_pre_ping': True,
'pool_recycle': 300,
}
# JWT
JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY', 'jwt-secret-key-change-in-production')
JWT_ACCESS_TOKEN_EXPIRES = timedelta(
seconds=int(os.environ.get('JWT_ACCESS_TOKEN_EXPIRES', 3600))
)
JWT_REFRESH_TOKEN_EXPIRES = timedelta(
seconds=int(os.environ.get('JWT_REFRESH_TOKEN_EXPIRES', 2592000))
)
# CORS
CORS_ORIGINS = os.environ.get('CORS_ORIGINS', '*').split(',')
# Logging
LOG_LEVEL = os.environ.get('LOG_LEVEL', 'INFO')
# Pagination
DEFAULT_PAGE_SIZE = 20
MAX_PAGE_SIZE = 100
class DevelopmentConfig(Config):
"""Development configuration."""
DEBUG = True
SQLALCHEMY_ECHO = True
# Use SQLite for local development if no DATABASE_URL set
SQLALCHEMY_DATABASE_URI = os.environ.get(
'DATABASE_URL',
'sqlite:///shopdb_dev.db'
)
SQLALCHEMY_ENGINE_OPTIONS = {} # SQLite doesn't need pool options
class TestingConfig(Config):
"""Testing configuration."""
TESTING = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
JWT_ACCESS_TOKEN_EXPIRES = timedelta(seconds=5)
class ProductionConfig(Config):
"""Production configuration."""
DEBUG = False
SQLALCHEMY_ECHO = False
# Stricter security in production
JWT_COOKIE_SECURE = True
JWT_COOKIE_CSRF_PROTECT = True
config = {
'development': DevelopmentConfig,
'testing': TestingConfig,
'production': ProductionConfig,
'default': DevelopmentConfig
}

1
shopdb/core/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Core module - always loaded."""

View File

@@ -0,0 +1,33 @@
"""Core API blueprints."""
from .auth import auth_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
from .locations import locations_bp
from .operatingsystems import operatingsystems_bp
from .dashboard import dashboard_bp
from .applications import applications_bp
from .knowledgebase import knowledgebase_bp
from .search import search_bp
__all__ = [
'auth_bp',
'machines_bp',
'machinetypes_bp',
'pctypes_bp',
'statuses_bp',
'vendors_bp',
'models_bp',
'businessunits_bp',
'locations_bp',
'operatingsystems_bp',
'dashboard_bp',
'applications_bp',
'knowledgebase_bp',
'search_bp',
]

View File

@@ -0,0 +1,429 @@
"""Applications API endpoints."""
from flask import Blueprint, request
from flask_jwt_extended import jwt_required
from shopdb.extensions import db
from shopdb.core.models import (
Application, AppVersion, AppOwner, SupportTeam, InstalledApp, Machine
)
from shopdb.utils.responses import (
success_response,
error_response,
paginated_response,
ErrorCodes
)
from shopdb.utils.pagination import get_pagination_params, paginate_query
applications_bp = Blueprint('applications', __name__)
@applications_bp.route('', methods=['GET'])
@jwt_required(optional=True)
def list_applications():
"""List all applications."""
page, per_page = get_pagination_params(request)
query = Application.query
if request.args.get('active', 'true').lower() != 'false':
query = query.filter(Application.isactive == True)
# Filter out hidden unless specifically requested
if request.args.get('showhidden', 'false').lower() != 'true':
query = query.filter(Application.ishidden == False)
# Filter by installable
if request.args.get('installable') is not None:
installable = request.args.get('installable').lower() == 'true'
query = query.filter(Application.isinstallable == installable)
if search := request.args.get('search'):
query = query.filter(
db.or_(
Application.appname.ilike(f'%{search}%'),
Application.appdescription.ilike(f'%{search}%')
)
)
query = query.order_by(Application.appname)
items, total = paginate_query(query, page, per_page)
data = []
for app in items:
app_dict = app.to_dict()
if app.supportteam:
app_dict['supportteam'] = {
'supportteamid': app.supportteam.supportteamid,
'teamname': app.supportteam.teamname,
'teamurl': app.supportteam.teamurl,
'owner': {
'appownerid': app.supportteam.owner.appownerid,
'appowner': app.supportteam.owner.appowner,
'sso': app.supportteam.owner.sso
} if app.supportteam.owner else None
}
else:
app_dict['supportteam'] = None
app_dict['installedcount'] = app.installed_on.filter_by(isactive=True).count()
data.append(app_dict)
return paginated_response(data, page, per_page, total)
@applications_bp.route('/<int:app_id>', methods=['GET'])
@jwt_required(optional=True)
def get_application(app_id: int):
"""Get a single application with details."""
app = Application.query.get(app_id)
if not app:
return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404)
data = app.to_dict()
if app.supportteam:
data['supportteam'] = {
'supportteamid': app.supportteam.supportteamid,
'teamname': app.supportteam.teamname,
'teamurl': app.supportteam.teamurl,
'owner': {
'appownerid': app.supportteam.owner.appownerid,
'appowner': app.supportteam.owner.appowner,
'sso': app.supportteam.owner.sso
} if app.supportteam.owner else None
}
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()
return success_response(data)
@applications_bp.route('', methods=['POST'])
@jwt_required()
def create_application():
"""Create a new application."""
data = request.get_json()
if not data or not data.get('appname'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'appname is required')
if Application.query.filter_by(appname=data['appname']).first():
return error_response(
ErrorCodes.CONFLICT,
f"Application '{data['appname']}' already exists",
http_code=409
)
app = Application(
appname=data['appname'],
appdescription=data.get('appdescription'),
supportteamid=data.get('supportteamid'),
isinstallable=data.get('isinstallable', False),
applicationnotes=data.get('applicationnotes'),
installpath=data.get('installpath'),
applicationlink=data.get('applicationlink'),
documentationpath=data.get('documentationpath'),
ishidden=data.get('ishidden', False),
isprinter=data.get('isprinter', False),
islicenced=data.get('islicenced', False),
image=data.get('image')
)
db.session.add(app)
db.session.commit()
return success_response(app.to_dict(), message='Application created', http_code=201)
@applications_bp.route('/<int:app_id>', methods=['PUT'])
@jwt_required()
def update_application(app_id: int):
"""Update an application."""
app = Application.query.get(app_id)
if not app:
return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404)
data = request.get_json()
if not data:
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
if 'appname' in data and data['appname'] != app.appname:
if Application.query.filter_by(appname=data['appname']).first():
return error_response(
ErrorCodes.CONFLICT,
f"Application '{data['appname']}' already exists",
http_code=409
)
fields = [
'appname', 'appdescription', 'supportteamid', 'isinstallable',
'applicationnotes', 'installpath', 'applicationlink', 'documentationpath',
'ishidden', 'isprinter', 'islicenced', 'image', 'isactive'
]
for key in fields:
if key in data:
setattr(app, key, data[key])
db.session.commit()
return success_response(app.to_dict(), message='Application updated')
@applications_bp.route('/<int:app_id>', methods=['DELETE'])
@jwt_required()
def delete_application(app_id: int):
"""Delete (deactivate) an application."""
app = Application.query.get(app_id)
if not app:
return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404)
app.isactive = False
db.session.commit()
return success_response(message='Application deleted')
# ---- Versions ----
@applications_bp.route('/<int:app_id>/versions', methods=['GET'])
@jwt_required(optional=True)
def list_versions(app_id: int):
"""List all versions for an application."""
app = Application.query.get(app_id)
if not app:
return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404)
versions = app.versions.filter_by(isactive=True).order_by(AppVersion.version.desc()).all()
return success_response([v.to_dict() for v in versions])
@applications_bp.route('/<int:app_id>/versions', methods=['POST'])
@jwt_required()
def create_version(app_id: int):
"""Create a new version for an application."""
app = Application.query.get(app_id)
if not app:
return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404)
data = request.get_json()
if not data or not data.get('version'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'version is required')
if AppVersion.query.filter_by(appid=app_id, version=data['version']).first():
return error_response(
ErrorCodes.CONFLICT,
f"Version '{data['version']}' already exists for this application",
http_code=409
)
version = AppVersion(
appid=app_id,
version=data['version'],
releasedate=data.get('releasedate'),
notes=data.get('notes')
)
db.session.add(version)
db.session.commit()
return success_response(version.to_dict(), message='Version created', http_code=201)
# ---- Machines with this app installed ----
@applications_bp.route('/<int:app_id>/installed', methods=['GET'])
@jwt_required(optional=True)
def list_installed_machines(app_id: int):
"""List all machines that have this application installed."""
app = Application.query.get(app_id)
if not app:
return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404)
installed = app.installed_on.filter_by(isactive=True).all()
data = []
for i in installed:
item = i.to_dict()
if i.machine:
item['machine'] = {
'machineid': i.machine.machineid,
'machinenumber': i.machine.machinenumber,
'alias': i.machine.alias,
'hostname': i.machine.hostname
}
data.append(item)
return success_response(data)
# ---- Installed Apps (per machine) ----
@applications_bp.route('/machines/<int:machine_id>', methods=['GET'])
@jwt_required(optional=True)
def list_machine_applications(machine_id: int):
"""List all applications installed on a machine."""
machine = Machine.query.get(machine_id)
if not machine:
return error_response(ErrorCodes.NOT_FOUND, 'Machine not found', http_code=404)
installed = machine.installedapps.filter_by(isactive=True).all()
return success_response([i.to_dict() for i in installed])
@applications_bp.route('/machines/<int:machine_id>', methods=['POST'])
@jwt_required()
def install_application(machine_id: int):
"""Install an application on a machine."""
machine = Machine.query.get(machine_id)
if not machine:
return error_response(ErrorCodes.NOT_FOUND, 'Machine not found', http_code=404)
data = request.get_json()
if not data or not data.get('appid'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'appid is required')
app = Application.query.get(data['appid'])
if not app:
return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404)
# Check if already installed
existing = InstalledApp.query.filter_by(
machineid=machine_id,
appid=data['appid']
).first()
if existing:
if existing.isactive:
return error_response(
ErrorCodes.CONFLICT,
'Application already installed on this machine',
http_code=409
)
# Reactivate
existing.isactive = True
existing.appversionid = data.get('appversionid')
existing.installeddate = db.func.now()
db.session.commit()
return success_response(existing.to_dict(), message='Application reinstalled')
installed = InstalledApp(
machineid=machine_id,
appid=data['appid'],
appversionid=data.get('appversionid')
)
db.session.add(installed)
db.session.commit()
return success_response(installed.to_dict(), message='Application installed', http_code=201)
@applications_bp.route('/machines/<int:machine_id>/<int:app_id>', methods=['DELETE'])
@jwt_required()
def uninstall_application(machine_id: int, app_id: int):
"""Uninstall an application from a machine."""
installed = InstalledApp.query.filter_by(
machineid=machine_id,
appid=app_id,
isactive=True
).first()
if not installed:
return error_response(ErrorCodes.NOT_FOUND, 'Application not installed on this machine', http_code=404)
installed.isactive = False
db.session.commit()
return success_response(message='Application uninstalled')
@applications_bp.route('/machines/<int:machine_id>/<int:app_id>', methods=['PUT'])
@jwt_required()
def update_installed_app(machine_id: int, app_id: int):
"""Update installed application (e.g., change version)."""
installed = InstalledApp.query.filter_by(
machineid=machine_id,
appid=app_id,
isactive=True
).first()
if not installed:
return error_response(ErrorCodes.NOT_FOUND, 'Application not installed on this machine', http_code=404)
data = request.get_json()
if not data:
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
if 'appversionid' in data:
installed.appversionid = data['appversionid']
db.session.commit()
return success_response(installed.to_dict(), message='Installation updated')
# ---- Support Teams ----
@applications_bp.route('/supportteams', methods=['GET'])
@jwt_required(optional=True)
def list_support_teams():
"""List all support teams."""
teams = SupportTeam.query.filter_by(isactive=True).order_by(SupportTeam.teamname).all()
data = []
for team in teams:
team_dict = team.to_dict()
team_dict['owner'] = team.owner.appowner if team.owner else None
data.append(team_dict)
return success_response(data)
@applications_bp.route('/supportteams', methods=['POST'])
@jwt_required()
def create_support_team():
"""Create a new support team."""
data = request.get_json()
if not data or not data.get('teamname'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'teamname is required')
team = SupportTeam(
teamname=data['teamname'],
teamurl=data.get('teamurl'),
appownerid=data.get('appownerid')
)
db.session.add(team)
db.session.commit()
return success_response(team.to_dict(), message='Support team created', http_code=201)
# ---- App Owners ----
@applications_bp.route('/appowners', methods=['GET'])
@jwt_required(optional=True)
def list_app_owners():
"""List all application owners."""
owners = AppOwner.query.filter_by(isactive=True).order_by(AppOwner.appowner).all()
return success_response([o.to_dict() for o in owners])
@applications_bp.route('/appowners', methods=['POST'])
@jwt_required()
def create_app_owner():
"""Create a new application owner."""
data = request.get_json()
if not data or not data.get('appowner'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'appowner is required')
owner = AppOwner(
appowner=data['appowner'],
sso=data.get('sso'),
email=data.get('email')
)
db.session.add(owner)
db.session.commit()
return success_response(owner.to_dict(), message='App owner created', http_code=201)

147
shopdb/core/api/auth.py Normal file
View File

@@ -0,0 +1,147 @@
"""Authentication API endpoints."""
from flask import Blueprint, request
from flask_jwt_extended import (
create_access_token,
create_refresh_token,
jwt_required,
get_jwt_identity,
current_user
)
from werkzeug.security import check_password_hash
from shopdb.extensions import db
from shopdb.core.models import User
from shopdb.utils.responses import success_response, error_response, ErrorCodes
auth_bp = Blueprint('auth', __name__)
@auth_bp.route('/login', methods=['POST'])
def login():
"""
Authenticate user and return JWT tokens.
Request:
{
"username": "string",
"password": "string"
}
Response:
{
"data": {
"access_token": "...",
"refresh_token": "...",
"user": {...}
}
}
"""
data = request.get_json()
if not data or not data.get('username') or not data.get('password'):
return error_response(
ErrorCodes.VALIDATION_ERROR,
'Username and password required'
)
user = User.query.filter_by(
username=data['username'],
isactive=True
).first()
if not user or not check_password_hash(user.passwordhash, data['password']):
return error_response(
ErrorCodes.UNAUTHORIZED,
'Invalid username or password',
http_code=401
)
if user.islocked:
return error_response(
ErrorCodes.FORBIDDEN,
'Account is locked',
http_code=403
)
# Create tokens (identity must be a string in Flask-JWT-Extended 4.x)
access_token = create_access_token(
identity=str(user.userid),
additional_claims={
'username': user.username,
'roles': [r.rolename for r in user.roles]
}
)
refresh_token = create_refresh_token(identity=str(user.userid))
# Update last login
user.lastlogindate = db.func.now()
user.failedlogins = 0
db.session.commit()
return success_response({
'access_token': access_token,
'refresh_token': refresh_token,
'token_type': 'Bearer',
'expires_in': 3600,
'user': {
'userid': user.userid,
'username': user.username,
'email': user.email,
'firstname': user.firstname,
'lastname': user.lastname,
'roles': [r.rolename for r in user.roles]
}
})
@auth_bp.route('/refresh', methods=['POST'])
@jwt_required(refresh=True)
def refresh():
"""Refresh access token using refresh token."""
user_id = get_jwt_identity()
user = User.query.get(int(user_id))
if not user or not user.isactive:
return error_response(
ErrorCodes.UNAUTHORIZED,
'User not found or inactive',
http_code=401
)
access_token = create_access_token(
identity=str(user.userid),
additional_claims={
'username': user.username,
'roles': [r.rolename for r in user.roles]
}
)
return success_response({
'access_token': access_token,
'token_type': 'Bearer',
'expires_in': 3600
})
@auth_bp.route('/me', methods=['GET'])
@jwt_required()
def get_current_user():
"""Get current authenticated user info."""
return success_response({
'userid': current_user.userid,
'username': current_user.username,
'email': current_user.email,
'firstname': current_user.firstname,
'lastname': current_user.lastname,
'roles': [r.rolename for r in current_user.roles],
'permissions': current_user.getpermissions()
})
@auth_bp.route('/logout', methods=['POST'])
@jwt_required()
def logout():
"""Logout user (for frontend token cleanup)."""
# In a full implementation, you'd blacklist the token
return success_response(message='Successfully logged out')

View File

@@ -0,0 +1,144 @@
"""Business Units 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 BusinessUnit
from shopdb.utils.responses import (
success_response,
error_response,
paginated_response,
ErrorCodes
)
from shopdb.utils.pagination import get_pagination_params, paginate_query
businessunits_bp = Blueprint('businessunits', __name__)
@businessunits_bp.route('', methods=['GET'])
@jwt_required(optional=True)
def list_businessunits():
"""List all business units."""
page, per_page = get_pagination_params(request)
query = BusinessUnit.query
if request.args.get('active', 'true').lower() != 'false':
query = query.filter(BusinessUnit.isactive == True)
if search := request.args.get('search'):
query = query.filter(
db.or_(
BusinessUnit.businessunit.ilike(f'%{search}%'),
BusinessUnit.code.ilike(f'%{search}%')
)
)
query = query.order_by(BusinessUnit.businessunit)
items, total = paginate_query(query, page, per_page)
data = [bu.to_dict() for bu in items]
return paginated_response(data, page, per_page, total)
@businessunits_bp.route('/<int:bu_id>', methods=['GET'])
@jwt_required(optional=True)
def get_businessunit(bu_id: int):
"""Get a single business unit."""
bu = BusinessUnit.query.get(bu_id)
if not bu:
return error_response(
ErrorCodes.NOT_FOUND,
f'Business unit with ID {bu_id} not found',
http_code=404
)
data = bu.to_dict()
data['parent'] = bu.parent.to_dict() if bu.parent else None
data['children'] = [c.to_dict() for c in bu.children]
return success_response(data)
@businessunits_bp.route('', methods=['POST'])
@jwt_required()
def create_businessunit():
"""Create a new business unit."""
data = request.get_json()
if not data or not data.get('businessunit'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'businessunit is required')
if BusinessUnit.query.filter_by(businessunit=data['businessunit']).first():
return error_response(
ErrorCodes.CONFLICT,
f"Business unit '{data['businessunit']}' already exists",
http_code=409
)
bu = BusinessUnit(
businessunit=data['businessunit'],
code=data.get('code'),
description=data.get('description'),
parentid=data.get('parentid')
)
db.session.add(bu)
db.session.commit()
return success_response(bu.to_dict(), message='Business unit created', http_code=201)
@businessunits_bp.route('/<int:bu_id>', methods=['PUT'])
@jwt_required()
def update_businessunit(bu_id: int):
"""Update a business unit."""
bu = BusinessUnit.query.get(bu_id)
if not bu:
return error_response(
ErrorCodes.NOT_FOUND,
f'Business unit with ID {bu_id} not found',
http_code=404
)
data = request.get_json()
if not data:
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
if 'businessunit' in data and data['businessunit'] != bu.businessunit:
if BusinessUnit.query.filter_by(businessunit=data['businessunit']).first():
return error_response(
ErrorCodes.CONFLICT,
f"Business unit '{data['businessunit']}' already exists",
http_code=409
)
for key in ['businessunit', 'code', 'description', 'parentid', 'isactive']:
if key in data:
setattr(bu, key, data[key])
db.session.commit()
return success_response(bu.to_dict(), message='Business unit updated')
@businessunits_bp.route('/<int:bu_id>', methods=['DELETE'])
@jwt_required()
def delete_businessunit(bu_id: int):
"""Delete (deactivate) a business unit."""
bu = BusinessUnit.query.get(bu_id)
if not bu:
return error_response(
ErrorCodes.NOT_FOUND,
f'Business unit with ID {bu_id} not found',
http_code=404
)
bu.isactive = False
db.session.commit()
return success_response(message='Business unit deleted')

View File

@@ -0,0 +1,117 @@
"""Dashboard API endpoints."""
from flask import Blueprint
from flask_jwt_extended import jwt_required
from shopdb.extensions import db
from shopdb.core.models import Machine, MachineType, MachineStatus
from shopdb.utils.responses import success_response
dashboard_bp = Blueprint('dashboard', __name__)
@dashboard_bp.route('/summary', methods=['GET'])
@dashboard_bp.route('', methods=['GET'])
@jwt_required(optional=True)
def get_dashboard():
"""Get dashboard summary data."""
# Count machines by category
equipment_count = db.session.query(Machine).join(MachineType).filter(
Machine.isactive == True,
MachineType.category == 'Equipment'
).count()
pc_count = db.session.query(Machine).join(MachineType).filter(
Machine.isactive == True,
MachineType.category == 'PC'
).count()
network_count = db.session.query(Machine).join(MachineType).filter(
Machine.isactive == True,
MachineType.category == 'Network'
).count()
# Count by status
status_counts = db.session.query(
MachineStatus.status,
db.func.count(Machine.machineid)
).outerjoin(
Machine,
db.and_(Machine.statusid == MachineStatus.statusid, Machine.isactive == True)
).group_by(MachineStatus.status).all()
# Recent machines
recent_machines = Machine.query.filter_by(isactive=True).order_by(
Machine.createddate.desc()
).limit(10).all()
# Build status dict
status_dict = {status: count for status, count in status_counts}
return success_response({
# Fields expected by frontend
'totalmachines': equipment_count + pc_count + network_count,
'totalequipment': equipment_count,
'totalpc': pc_count,
'totalnetwork': network_count,
'activemachines': status_dict.get('In Use', 0),
'inrepair': status_dict.get('In Repair', 0),
# Also include structured data
'counts': {
'equipment': equipment_count,
'pcs': pc_count,
'network_devices': network_count,
'total': equipment_count + pc_count + network_count
},
'by_status': status_dict,
'recent': [
{
'machineid': m.machineid,
'machinenumber': m.machinenumber,
'machinetype': m.machinetype.machinetype if m.machinetype else None,
'createddate': m.createddate.isoformat() + 'Z' if m.createddate else None
}
for m in recent_machines
]
})
@dashboard_bp.route('/stats', methods=['GET'])
@jwt_required(optional=True)
def get_stats():
"""Get detailed statistics."""
# Machine type breakdown
type_counts = db.session.query(
MachineType.machinetype,
MachineType.category,
db.func.count(Machine.machineid)
).outerjoin(
Machine,
db.and_(Machine.machinetypeid == MachineType.machinetypeid, Machine.isactive == True)
).filter(MachineType.isactive == True).group_by(
MachineType.machinetypeid
).all()
return success_response({
'by_type': [
{'type': t, 'category': c, 'count': count}
for t, c, count in type_counts
]
})
@dashboard_bp.route('/health', methods=['GET'])
def health_check():
"""Health check endpoint (no auth required)."""
try:
# Test database connection
db.session.execute(db.text('SELECT 1'))
db_status = 'healthy'
except Exception as e:
db_status = f'unhealthy: {str(e)}'
return success_response({
'status': 'ok' if db_status == 'healthy' else 'degraded',
'database': db_status,
'version': '1.0.0'
})

View File

@@ -0,0 +1,207 @@
"""Knowledge Base API endpoints."""
from flask import Blueprint, request
from flask_jwt_extended import jwt_required
from shopdb.extensions import db
from shopdb.core.models import KnowledgeBase, Application
from shopdb.utils.responses import (
success_response,
error_response,
paginated_response,
ErrorCodes
)
from shopdb.utils.pagination import get_pagination_params, paginate_query
knowledgebase_bp = Blueprint('knowledgebase', __name__)
@knowledgebase_bp.route('', methods=['GET'])
@jwt_required(optional=True)
def list_articles():
"""List all knowledge base articles."""
page, per_page = get_pagination_params(request)
query = KnowledgeBase.query.filter_by(isactive=True)
# Search
if search := request.args.get('search'):
query = query.filter(
db.or_(
KnowledgeBase.shortdescription.ilike(f'%{search}%'),
KnowledgeBase.keywords.ilike(f'%{search}%')
)
)
# Filter by topic/application
if appid := request.args.get('appid'):
query = query.filter(KnowledgeBase.appid == int(appid))
# Sort options
sort = request.args.get('sort', 'clicks')
order = request.args.get('order', 'desc')
if sort == 'clicks':
query = query.order_by(
KnowledgeBase.clicks.desc() if order == 'desc' else KnowledgeBase.clicks.asc(),
KnowledgeBase.lastupdated.desc()
)
elif sort == 'topic':
query = query.join(Application).order_by(
Application.appname.desc() if order == 'desc' else Application.appname.asc()
)
elif sort == 'description':
query = query.order_by(
KnowledgeBase.shortdescription.desc() if order == 'desc' else KnowledgeBase.shortdescription.asc()
)
elif sort == 'lastupdated':
query = query.order_by(
KnowledgeBase.lastupdated.desc() if order == 'desc' else KnowledgeBase.lastupdated.asc()
)
else:
query = query.order_by(KnowledgeBase.clicks.desc())
items, total = paginate_query(query, page, per_page)
data = []
for article in items:
article_dict = article.to_dict()
if article.application:
article_dict['application'] = {
'appid': article.application.appid,
'appname': article.application.appname
}
else:
article_dict['application'] = None
data.append(article_dict)
return paginated_response(data, page, per_page, total)
@knowledgebase_bp.route('/stats', methods=['GET'])
@jwt_required(optional=True)
def get_stats():
"""Get knowledge base statistics."""
total_clicks = db.session.query(
db.func.coalesce(db.func.sum(KnowledgeBase.clicks), 0)
).filter(KnowledgeBase.isactive == True).scalar()
total_articles = KnowledgeBase.query.filter_by(isactive=True).count()
return success_response({
'totalclicks': int(total_clicks),
'totalarticles': total_articles
})
@knowledgebase_bp.route('/<int:link_id>', methods=['GET'])
@jwt_required(optional=True)
def get_article(link_id: int):
"""Get a single knowledge base article."""
article = KnowledgeBase.query.get(link_id)
if not article or not article.isactive:
return error_response(ErrorCodes.NOT_FOUND, 'Article not found', http_code=404)
data = article.to_dict()
if article.application:
data['application'] = {
'appid': article.application.appid,
'appname': article.application.appname
}
else:
data['application'] = None
return success_response(data)
@knowledgebase_bp.route('/<int:link_id>/click', methods=['POST'])
@jwt_required(optional=True)
def track_click(link_id: int):
"""Increment click counter and return the URL to redirect to."""
article = KnowledgeBase.query.get(link_id)
if not article or not article.isactive:
return error_response(ErrorCodes.NOT_FOUND, 'Article not found', http_code=404)
article.increment_clicks()
db.session.commit()
return success_response({
'linkurl': article.linkurl,
'clicks': article.clicks
})
@knowledgebase_bp.route('', methods=['POST'])
@jwt_required()
def create_article():
"""Create a new knowledge base article."""
data = request.get_json()
if not data or not data.get('shortdescription'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'shortdescription is required')
if not data.get('linkurl'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'linkurl is required')
# Validate application if provided
if data.get('appid'):
app = Application.query.get(data['appid'])
if not app:
return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404)
article = KnowledgeBase(
shortdescription=data['shortdescription'],
linkurl=data['linkurl'],
appid=data.get('appid'),
keywords=data.get('keywords'),
clicks=0
)
db.session.add(article)
db.session.commit()
return success_response(article.to_dict(), message='Article created', http_code=201)
@knowledgebase_bp.route('/<int:link_id>', methods=['PUT'])
@jwt_required()
def update_article(link_id: int):
"""Update a knowledge base article."""
article = KnowledgeBase.query.get(link_id)
if not article:
return error_response(ErrorCodes.NOT_FOUND, 'Article not found', http_code=404)
data = request.get_json()
if not data:
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
# Validate application if being changed
if 'appid' in data and data['appid']:
app = Application.query.get(data['appid'])
if not app:
return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404)
fields = ['shortdescription', 'linkurl', 'appid', 'keywords', 'isactive']
for key in fields:
if key in data:
setattr(article, key, data[key])
db.session.commit()
return success_response(article.to_dict(), message='Article updated')
@knowledgebase_bp.route('/<int:link_id>', methods=['DELETE'])
@jwt_required()
def delete_article(link_id: int):
"""Delete (deactivate) a knowledge base article."""
article = KnowledgeBase.query.get(link_id)
if not article:
return error_response(ErrorCodes.NOT_FOUND, 'Article not found', http_code=404)
article.isactive = False
db.session.commit()
return success_response(message='Article deleted')

View File

@@ -0,0 +1,144 @@
"""Locations 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 Location
from shopdb.utils.responses import (
success_response,
error_response,
paginated_response,
ErrorCodes
)
from shopdb.utils.pagination import get_pagination_params, paginate_query
locations_bp = Blueprint('locations', __name__)
@locations_bp.route('', methods=['GET'])
@jwt_required()
def list_locations():
"""List all locations."""
page, per_page = get_pagination_params(request)
query = Location.query
if request.args.get('active', 'true').lower() != 'false':
query = query.filter(Location.isactive == True)
if search := request.args.get('search'):
query = query.filter(
db.or_(
Location.locationname.ilike(f'%{search}%'),
Location.building.ilike(f'%{search}%')
)
)
query = query.order_by(Location.locationname)
items, total = paginate_query(query, page, per_page)
data = [loc.to_dict() for loc in items]
return paginated_response(data, page, per_page, total)
@locations_bp.route('/<int:location_id>', methods=['GET'])
@jwt_required()
def get_location(location_id: int):
"""Get a single location."""
loc = Location.query.get(location_id)
if not loc:
return error_response(
ErrorCodes.NOT_FOUND,
f'Location with ID {location_id} not found',
http_code=404
)
return success_response(loc.to_dict())
@locations_bp.route('', methods=['POST'])
@jwt_required()
def create_location():
"""Create a new location."""
data = request.get_json()
if not data or not data.get('locationname'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'locationname is required')
if Location.query.filter_by(locationname=data['locationname']).first():
return error_response(
ErrorCodes.CONFLICT,
f"Location '{data['locationname']}' already exists",
http_code=409
)
loc = Location(
locationname=data['locationname'],
building=data.get('building'),
floor=data.get('floor'),
room=data.get('room'),
description=data.get('description'),
mapimage=data.get('mapimage'),
mapwidth=data.get('mapwidth'),
mapheight=data.get('mapheight')
)
db.session.add(loc)
db.session.commit()
return success_response(loc.to_dict(), message='Location created', http_code=201)
@locations_bp.route('/<int:location_id>', methods=['PUT'])
@jwt_required()
def update_location(location_id: int):
"""Update a location."""
loc = Location.query.get(location_id)
if not loc:
return error_response(
ErrorCodes.NOT_FOUND,
f'Location with ID {location_id} not found',
http_code=404
)
data = request.get_json()
if not data:
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
if 'locationname' in data and data['locationname'] != loc.locationname:
if Location.query.filter_by(locationname=data['locationname']).first():
return error_response(
ErrorCodes.CONFLICT,
f"Location '{data['locationname']}' already exists",
http_code=409
)
for key in ['locationname', 'building', 'floor', 'room', 'description', 'mapimage', 'mapwidth', 'mapheight', 'isactive']:
if key in data:
setattr(loc, key, data[key])
db.session.commit()
return success_response(loc.to_dict(), message='Location updated')
@locations_bp.route('/<int:location_id>', methods=['DELETE'])
@jwt_required()
def delete_location(location_id: int):
"""Delete (deactivate) a location."""
loc = Location.query.get(location_id)
if not loc:
return error_response(
ErrorCodes.NOT_FOUND,
f'Location with ID {location_id} not found',
http_code=404
)
loc.isactive = False
db.session.commit()
return success_response(message='Location deleted')

567
shopdb/core/api/machines.py Normal file
View File

@@ -0,0 +1,567 @@
"""Machines API endpoints."""
from flask import Blueprint, request
from flask_jwt_extended import jwt_required, current_user
from shopdb.extensions import db
from shopdb.core.models import Machine, MachineType
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
machines_bp = Blueprint('machines', __name__)
@machines_bp.route('', methods=['GET'])
@jwt_required(optional=True)
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)
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()
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.commit()
return success_response(
machine.to_dict(),
message='Machine created successfully',
http_code=201
)
@machines_bp.route('/<int:machine_id>', methods=['PUT'])
@jwt_required()
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'
]
for key, value in data.items():
if key in allowed_fields:
setattr(machine, key, value)
machine.modifiedby = current_user.username
db.session.commit()
return success_response(machine.to_dict(), message='Machine updated successfully')
@machines_bp.route('/<int:machine_id>', methods=['DELETE'])
@jwt_required()
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)
db.session.commit()
return success_response(message='Machine deleted successfully')
@machines_bp.route('/<int:machine_id>/communications', methods=['GET'])
@jwt_required()
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()
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)
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()
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()
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)
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()
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)

View File

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

151
shopdb/core/api/models.py Normal file
View File

@@ -0,0 +1,151 @@
"""Models (equipment models) 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 Model
from shopdb.utils.responses import (
success_response,
error_response,
paginated_response,
ErrorCodes
)
from shopdb.utils.pagination import get_pagination_params, paginate_query
models_bp = Blueprint('models', __name__)
@models_bp.route('', methods=['GET'])
@jwt_required(optional=True)
def list_models():
"""List all equipment models."""
page, per_page = get_pagination_params(request)
query = Model.query
if request.args.get('active', 'true').lower() != 'false':
query = query.filter(Model.isactive == True)
if vendor_id := request.args.get('vendor', type=int):
query = query.filter(Model.vendorid == vendor_id)
if machinetype_id := request.args.get('machinetype', type=int):
query = query.filter(Model.machinetypeid == machinetype_id)
if search := request.args.get('search'):
query = query.filter(Model.modelnumber.ilike(f'%{search}%'))
query = query.order_by(Model.modelnumber)
items, total = paginate_query(query, page, per_page)
data = []
for m in items:
d = m.to_dict()
d['vendor'] = m.vendor.vendor if m.vendor else None
d['machinetype'] = m.machinetype.machinetype if m.machinetype else None
data.append(d)
return paginated_response(data, page, per_page, total)
@models_bp.route('/<int:model_id>', methods=['GET'])
@jwt_required()
def get_model(model_id: int):
"""Get a single model."""
m = Model.query.get(model_id)
if not m:
return error_response(
ErrorCodes.NOT_FOUND,
f'Model with ID {model_id} not found',
http_code=404
)
data = m.to_dict()
data['vendor'] = m.vendor.to_dict() if m.vendor else None
data['machinetype'] = m.machinetype.to_dict() if m.machinetype else None
return success_response(data)
@models_bp.route('', methods=['POST'])
@jwt_required()
def create_model():
"""Create a new model."""
data = request.get_json()
if not data or not data.get('modelnumber'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'modelnumber is required')
# Check duplicate
existing = Model.query.filter_by(
modelnumber=data['modelnumber'],
vendorid=data.get('vendorid')
).first()
if existing:
return error_response(
ErrorCodes.CONFLICT,
f"Model '{data['modelnumber']}' already exists for this vendor",
http_code=409
)
m = Model(
modelnumber=data['modelnumber'],
vendorid=data.get('vendorid'),
machinetypeid=data.get('machinetypeid'),
description=data.get('description'),
imageurl=data.get('imageurl'),
documentationurl=data.get('documentationurl'),
notes=data.get('notes')
)
db.session.add(m)
db.session.commit()
return success_response(m.to_dict(), message='Model created', http_code=201)
@models_bp.route('/<int:model_id>', methods=['PUT'])
@jwt_required()
def update_model(model_id: int):
"""Update a model."""
m = Model.query.get(model_id)
if not m:
return error_response(
ErrorCodes.NOT_FOUND,
f'Model with ID {model_id} not found',
http_code=404
)
data = request.get_json()
if not data:
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
for key in ['modelnumber', 'vendorid', 'machinetypeid', 'description', 'imageurl', 'documentationurl', 'notes', 'isactive']:
if key in data:
setattr(m, key, data[key])
db.session.commit()
return success_response(m.to_dict(), message='Model updated')
@models_bp.route('/<int:model_id>', methods=['DELETE'])
@jwt_required()
def delete_model(model_id: int):
"""Delete (deactivate) a model."""
m = Model.query.get(model_id)
if not m:
return error_response(
ErrorCodes.NOT_FOUND,
f'Model with ID {model_id} not found',
http_code=404
)
m.isactive = False
db.session.commit()
return success_response(message='Model deleted')

View File

@@ -0,0 +1,131 @@
"""Operating Systems 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 OperatingSystem
from shopdb.utils.responses import (
success_response,
error_response,
paginated_response,
ErrorCodes
)
from shopdb.utils.pagination import get_pagination_params, paginate_query
operatingsystems_bp = Blueprint('operatingsystems', __name__)
@operatingsystems_bp.route('', methods=['GET'])
@jwt_required()
def list_operatingsystems():
"""List all operating systems."""
page, per_page = get_pagination_params(request)
query = OperatingSystem.query
if request.args.get('active', 'true').lower() != 'false':
query = query.filter(OperatingSystem.isactive == True)
if search := request.args.get('search'):
query = query.filter(OperatingSystem.osname.ilike(f'%{search}%'))
query = query.order_by(OperatingSystem.osname)
items, total = paginate_query(query, page, per_page)
data = [os.to_dict() for os in items]
return paginated_response(data, page, per_page, total)
@operatingsystems_bp.route('/<int:os_id>', methods=['GET'])
@jwt_required()
def get_operatingsystem(os_id: int):
"""Get a single operating system."""
os = OperatingSystem.query.get(os_id)
if not os:
return error_response(
ErrorCodes.NOT_FOUND,
f'Operating system with ID {os_id} not found',
http_code=404
)
return success_response(os.to_dict())
@operatingsystems_bp.route('', methods=['POST'])
@jwt_required()
def create_operatingsystem():
"""Create a new operating system."""
data = request.get_json()
if not data or not data.get('osname'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'osname is required')
existing = OperatingSystem.query.filter_by(
osname=data['osname'],
osversion=data.get('osversion')
).first()
if existing:
return error_response(
ErrorCodes.CONFLICT,
f"Operating system '{data['osname']} {data.get('osversion', '')}' already exists",
http_code=409
)
os = OperatingSystem(
osname=data['osname'],
osversion=data.get('osversion'),
architecture=data.get('architecture'),
endoflife=data.get('endoflife')
)
db.session.add(os)
db.session.commit()
return success_response(os.to_dict(), message='Operating system created', http_code=201)
@operatingsystems_bp.route('/<int:os_id>', methods=['PUT'])
@jwt_required()
def update_operatingsystem(os_id: int):
"""Update an operating system."""
os = OperatingSystem.query.get(os_id)
if not os:
return error_response(
ErrorCodes.NOT_FOUND,
f'Operating system with ID {os_id} not found',
http_code=404
)
data = request.get_json()
if not data:
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
for key in ['osname', 'osversion', 'architecture', 'endoflife', 'isactive']:
if key in data:
setattr(os, key, data[key])
db.session.commit()
return success_response(os.to_dict(), message='Operating system updated')
@operatingsystems_bp.route('/<int:os_id>', methods=['DELETE'])
@jwt_required()
def delete_operatingsystem(os_id: int):
"""Delete (deactivate) an operating system."""
os = OperatingSystem.query.get(os_id)
if not os:
return error_response(
ErrorCodes.NOT_FOUND,
f'Operating system with ID {os_id} not found',
http_code=404
)
os.isactive = False
db.session.commit()
return success_response(message='Operating system deleted')

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

@@ -0,0 +1,141 @@
"""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()
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()
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')

203
shopdb/core/api/search.py Normal file
View File

@@ -0,0 +1,203 @@
"""Global search API endpoint."""
from flask import Blueprint, request
from flask_jwt_extended import jwt_required
from shopdb.extensions import db
from shopdb.core.models import (
Machine, Application, KnowledgeBase
)
from shopdb.utils.responses import success_response
search_bp = Blueprint('search', __name__)
@search_bp.route('', methods=['GET'])
@jwt_required(optional=True)
def global_search():
"""
Global search across multiple entity types.
Returns combined results from:
- Machines (equipment and PCs)
- Applications
- Knowledge Base articles
- Printers (if available)
Results are sorted by relevance score.
"""
query = request.args.get('q', '').strip()
if not query or len(query) < 2:
return success_response({
'results': [],
'query': query,
'message': 'Search query must be at least 2 characters'
})
if len(query) > 200:
return success_response({
'results': [],
'query': query[:200],
'message': 'Search query too long'
})
results = []
search_term = f'%{query}%'
# Search Machines (Equipment and PCs)
machines = Machine.query.filter(
Machine.isactive == True,
db.or_(
Machine.machinenumber.ilike(search_term),
Machine.alias.ilike(search_term),
Machine.hostname.ilike(search_term),
Machine.serialnumber.ilike(search_term),
Machine.notes.ilike(search_term)
)
).limit(10).all()
for m in machines:
# Determine type: PC, Printer, or Equipment
is_pc = m.pctypeid is not None
is_printer = m.is_printer
# Calculate relevance - exact matches score higher
relevance = 15
if m.machinenumber and query.lower() == m.machinenumber.lower():
relevance = 100
elif m.hostname and query.lower() == m.hostname.lower():
relevance = 100
elif m.alias and query.lower() in m.alias.lower():
relevance = 50
display_name = m.hostname if is_pc and m.hostname else m.machinenumber
if m.alias and not is_pc:
display_name = f"{m.machinenumber} ({m.alias})"
# Determine result type and URL
if is_printer:
result_type = 'printer'
url = f"/printers/{m.machineid}"
elif is_pc:
result_type = 'pc'
url = f"/pcs/{m.machineid}"
else:
result_type = 'machine'
url = f"/machines/{m.machineid}"
# Get location - prefer machine's own location, fall back to parent machine's location
location_name = None
if m.location:
location_name = m.location.locationname
elif m.parent_relationships:
# Check parent machines for location
for rel in m.parent_relationships:
if rel.parent_machine and rel.parent_machine.location:
location_name = rel.parent_machine.location.locationname
break
# Get machinetype from model (single source of truth)
mt = m.derived_machinetype
results.append({
'type': result_type,
'id': m.machineid,
'title': display_name,
'subtitle': mt.machinetype if mt else None,
'location': location_name,
'url': url,
'relevance': relevance
})
# Search Applications
apps = Application.query.filter(
Application.isactive == True,
db.or_(
Application.appname.ilike(search_term),
Application.appdescription.ilike(search_term)
)
).limit(10).all()
for app in apps:
relevance = 20
if query.lower() == app.appname.lower():
relevance = 100
elif query.lower() in app.appname.lower():
relevance = 50
results.append({
'type': 'application',
'id': app.appid,
'title': app.appname,
'subtitle': app.appdescription[:100] if app.appdescription else None,
'url': f"/applications/{app.appid}",
'relevance': relevance
})
# Search Knowledge Base
kb_articles = KnowledgeBase.query.filter(
KnowledgeBase.isactive == True,
db.or_(
KnowledgeBase.shortdescription.ilike(search_term),
KnowledgeBase.keywords.ilike(search_term)
)
).limit(20).all()
for kb in kb_articles:
# Weight by clicks and keyword match
relevance = 10 + (kb.clicks or 0) * 0.1
if kb.keywords and query.lower() in kb.keywords.lower():
relevance += 15
results.append({
'type': 'knowledgebase',
'id': kb.linkid,
'title': kb.shortdescription,
'subtitle': kb.application.appname if kb.application else None,
'url': f"/knowledgebase/{kb.linkid}",
'linkurl': kb.linkurl,
'relevance': relevance
})
# Search Printers (check if printers model exists)
try:
from shopdb.plugins.printers.models import Printer
printers = Printer.query.filter(
Printer.isactive == True,
db.or_(
Printer.printercsfname.ilike(search_term),
Printer.printerwindowsname.ilike(search_term),
Printer.serialnumber.ilike(search_term),
Printer.fqdn.ilike(search_term)
)
).limit(10).all()
for p in printers:
relevance = 15
if p.printercsfname and query.lower() == p.printercsfname.lower():
relevance = 100
display_name = p.printercsfname or p.printerwindowsname or f"Printer #{p.printerid}"
results.append({
'type': 'printer',
'id': p.printerid,
'title': display_name,
'subtitle': p.printerwindowsname if p.printercsfname else None,
'url': f"/printers/{p.printerid}",
'relevance': relevance
})
except ImportError:
pass # Printers plugin not installed
# Sort by relevance (highest first)
results.sort(key=lambda x: x['relevance'], reverse=True)
# Limit total results
results = results[:30]
return success_response({
'results': results,
'query': query,
'total': len(results)
})

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

@@ -0,0 +1,139 @@
"""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')

137
shopdb/core/api/vendors.py Normal file
View File

@@ -0,0 +1,137 @@
"""Vendors 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 Vendor
from shopdb.utils.responses import (
success_response,
error_response,
paginated_response,
ErrorCodes
)
from shopdb.utils.pagination import get_pagination_params, paginate_query
vendors_bp = Blueprint('vendors', __name__)
@vendors_bp.route('', methods=['GET'])
@jwt_required()
def list_vendors():
"""List all vendors."""
page, per_page = get_pagination_params(request)
query = Vendor.query
if request.args.get('active', 'true').lower() != 'false':
query = query.filter(Vendor.isactive == True)
if search := request.args.get('search'):
query = query.filter(Vendor.vendor.ilike(f'%{search}%'))
query = query.order_by(Vendor.vendor)
items, total = paginate_query(query, page, per_page)
data = [v.to_dict() for v in items]
return paginated_response(data, page, per_page, total)
@vendors_bp.route('/<int:vendor_id>', methods=['GET'])
@jwt_required()
def get_vendor(vendor_id: int):
"""Get a single vendor."""
v = Vendor.query.get(vendor_id)
if not v:
return error_response(
ErrorCodes.NOT_FOUND,
f'Vendor with ID {vendor_id} not found',
http_code=404
)
return success_response(v.to_dict())
@vendors_bp.route('', methods=['POST'])
@jwt_required()
def create_vendor():
"""Create a new vendor."""
data = request.get_json()
if not data or not data.get('vendor'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'vendor is required')
if Vendor.query.filter_by(vendor=data['vendor']).first():
return error_response(
ErrorCodes.CONFLICT,
f"Vendor '{data['vendor']}' already exists",
http_code=409
)
v = Vendor(
vendor=data['vendor'],
description=data.get('description'),
website=data.get('website'),
supportphone=data.get('supportphone'),
supportemail=data.get('supportemail'),
notes=data.get('notes')
)
db.session.add(v)
db.session.commit()
return success_response(v.to_dict(), message='Vendor created', http_code=201)
@vendors_bp.route('/<int:vendor_id>', methods=['PUT'])
@jwt_required()
def update_vendor(vendor_id: int):
"""Update a vendor."""
v = Vendor.query.get(vendor_id)
if not v:
return error_response(
ErrorCodes.NOT_FOUND,
f'Vendor with ID {vendor_id} not found',
http_code=404
)
data = request.get_json()
if not data:
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
if 'vendor' in data and data['vendor'] != v.vendor:
if Vendor.query.filter_by(vendor=data['vendor']).first():
return error_response(
ErrorCodes.CONFLICT,
f"Vendor '{data['vendor']}' already exists",
http_code=409
)
for key in ['vendor', 'description', 'website', 'supportphone', 'supportemail', 'notes', 'isactive']:
if key in data:
setattr(v, key, data[key])
db.session.commit()
return success_response(v.to_dict(), message='Vendor updated')
@vendors_bp.route('/<int:vendor_id>', methods=['DELETE'])
@jwt_required()
def delete_vendor(vendor_id: int):
"""Delete (deactivate) a vendor."""
v = Vendor.query.get(vendor_id)
if not v:
return error_response(
ErrorCodes.NOT_FOUND,
f'Vendor with ID {vendor_id} not found',
http_code=404
)
v.isactive = False
db.session.commit()
return success_response(message='Vendor deleted')

View File

@@ -0,0 +1,49 @@
"""Core SQLAlchemy models."""
from .base import BaseModel, SoftDeleteMixin, AuditMixin
from .machine import Machine, MachineType, MachineStatus, PCType
from .vendor import Vendor
from .model import Model
from .businessunit import BusinessUnit
from .location import Location
from .operatingsystem import OperatingSystem
from .relationship import MachineRelationship, RelationshipType
from .communication import Communication, CommunicationType
from .user import User, Role
from .application import Application, AppVersion, AppOwner, SupportTeam, InstalledApp
from .knowledgebase import KnowledgeBase
__all__ = [
# Base
'BaseModel',
'SoftDeleteMixin',
'AuditMixin',
# Machine
'Machine',
'MachineType',
'MachineStatus',
'PCType',
# Reference
'Vendor',
'Model',
'BusinessUnit',
'Location',
'OperatingSystem',
# Relationships
'MachineRelationship',
'RelationshipType',
# Communication
'Communication',
'CommunicationType',
# Auth
'User',
'Role',
# Applications
'Application',
'AppVersion',
'AppOwner',
'SupportTeam',
'InstalledApp',
# Knowledge Base
'KnowledgeBase',
]

View File

@@ -0,0 +1,130 @@
"""Application tracking models."""
from shopdb.extensions import db
from .base import BaseModel
class AppOwner(BaseModel):
"""Application owner/contact."""
__tablename__ = 'appowners'
appownerid = db.Column(db.Integer, primary_key=True)
appowner = db.Column(db.String(100), nullable=False)
sso = db.Column(db.String(50))
email = db.Column(db.String(100))
# Relationships
supportteams = db.relationship('SupportTeam', back_populates='owner', lazy='dynamic')
def __repr__(self):
return f"<AppOwner {self.appowner}>"
class SupportTeam(BaseModel):
"""Application support team."""
__tablename__ = 'supportteams'
supportteamid = db.Column(db.Integer, primary_key=True)
teamname = db.Column(db.String(100), nullable=False)
teamurl = db.Column(db.String(255))
appownerid = db.Column(db.Integer, db.ForeignKey('appowners.appownerid'))
# Relationships
owner = db.relationship('AppOwner', back_populates='supportteams')
applications = db.relationship('Application', back_populates='supportteam', lazy='dynamic')
def __repr__(self):
return f"<SupportTeam {self.teamname}>"
class Application(BaseModel):
"""Application catalog."""
__tablename__ = 'applications'
appid = db.Column(db.Integer, primary_key=True)
appname = db.Column(db.String(100), unique=True, nullable=False)
appdescription = db.Column(db.String(255))
supportteamid = db.Column(db.Integer, db.ForeignKey('supportteams.supportteamid'))
isinstallable = db.Column(db.Boolean, default=False)
applicationnotes = db.Column(db.Text)
installpath = db.Column(db.String(255))
applicationlink = db.Column(db.String(512))
documentationpath = db.Column(db.String(512))
ishidden = db.Column(db.Boolean, default=False)
isprinter = db.Column(db.Boolean, default=False)
islicenced = db.Column(db.Boolean, default=False)
image = db.Column(db.String(255))
# Relationships
supportteam = db.relationship('SupportTeam', back_populates='applications')
versions = db.relationship('AppVersion', back_populates='application', lazy='dynamic')
installed_on = db.relationship('InstalledApp', back_populates='application', lazy='dynamic')
def __repr__(self):
return f"<Application {self.appname}>"
class AppVersion(BaseModel):
"""Application version tracking."""
__tablename__ = 'appversions'
appversionid = db.Column(db.Integer, primary_key=True)
appid = db.Column(db.Integer, db.ForeignKey('applications.appid'), nullable=False)
version = db.Column(db.String(50), nullable=False)
releasedate = db.Column(db.Date)
notes = db.Column(db.String(255))
dateadded = db.Column(db.DateTime, default=db.func.now())
# Relationships
application = db.relationship('Application', back_populates='versions')
installations = db.relationship('InstalledApp', back_populates='appversion', lazy='dynamic')
# Unique constraint on app + version
__table_args__ = (
db.UniqueConstraint('appid', 'version', name='uq_app_version'),
)
def __repr__(self):
return f"<AppVersion {self.application.appname if self.application else self.appid} v{self.version}>"
class InstalledApp(db.Model):
"""Junction table for applications installed on machines (PCs)."""
__tablename__ = 'installedapps'
id = db.Column(db.Integer, primary_key=True)
machineid = db.Column(db.Integer, db.ForeignKey('machines.machineid'), nullable=False)
appid = db.Column(db.Integer, db.ForeignKey('applications.appid'), nullable=False)
appversionid = db.Column(db.Integer, db.ForeignKey('appversions.appversionid'))
isactive = db.Column(db.Boolean, default=True, nullable=False)
installeddate = db.Column(db.DateTime, default=db.func.now())
# Relationships
machine = db.relationship('Machine', back_populates='installedapps')
application = db.relationship('Application', back_populates='installed_on')
appversion = db.relationship('AppVersion', back_populates='installations')
# Unique constraint - one app per machine (can have different versions over time)
__table_args__ = (
db.UniqueConstraint('machineid', 'appid', name='uq_machine_app'),
)
def to_dict(self):
"""Convert to dictionary."""
return {
'id': self.id,
'machineid': self.machineid,
'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"<InstalledApp machine={self.machineid} app={self.appid}>"

View File

@@ -0,0 +1,66 @@
"""Base model class with common fields."""
from datetime import datetime
from shopdb.extensions import db
class BaseModel(db.Model):
"""
Abstract base model with common fields.
All models should inherit from this.
"""
__abstract__ = True
createddate = db.Column(
db.DateTime,
default=datetime.utcnow,
nullable=False
)
modifieddate = db.Column(
db.DateTime,
default=datetime.utcnow,
onupdate=datetime.utcnow,
nullable=False
)
isactive = db.Column(db.Boolean, default=True, nullable=False)
def to_dict(self):
"""Convert model to dictionary."""
result = {}
for c in self.__table__.columns:
value = getattr(self, c.name)
if isinstance(value, datetime):
value = value.isoformat() + 'Z'
result[c.name] = value
return result
def update(self, **kwargs):
"""Update model attributes."""
for key, value in kwargs.items():
if hasattr(self, key):
setattr(self, key, value)
@classmethod
def get_active(cls):
"""Return query for active records only."""
return cls.query.filter_by(isactive=True)
class SoftDeleteMixin:
"""Mixin for soft delete functionality."""
deleteddate = db.Column(db.DateTime, nullable=True)
deletedby = db.Column(db.String(100), nullable=True)
def soft_delete(self, deleted_by: str = None):
"""Mark record as deleted."""
self.isactive = False
self.deleteddate = datetime.utcnow()
self.deletedby = deleted_by
class AuditMixin:
"""Mixin for audit fields."""
createdby = db.Column(db.String(100), nullable=True)
modifiedby = db.Column(db.String(100), nullable=True)

View File

@@ -0,0 +1,31 @@
"""Business Unit model."""
from shopdb.extensions import db
from .base import BaseModel
class BusinessUnit(BaseModel):
"""Business unit / department model."""
__tablename__ = 'businessunits'
businessunitid = db.Column(db.Integer, primary_key=True)
businessunit = db.Column(db.String(100), unique=True, nullable=False)
code = db.Column(db.String(20), unique=True, comment='Short code')
description = db.Column(db.Text)
# Optional parent for hierarchy
parentid = db.Column(
db.Integer,
db.ForeignKey('businessunits.businessunitid'),
nullable=True
)
# Self-referential relationship for hierarchy
parent = db.relationship(
'BusinessUnit',
remote_side=[businessunitid],
backref='children'
)
def __repr__(self):
return f"<BusinessUnit {self.businessunit}>"

View File

@@ -0,0 +1,90 @@
"""Communication/network interface models."""
from shopdb.extensions import db
from .base import BaseModel
class CommunicationType(BaseModel):
"""Types of communication interfaces."""
__tablename__ = 'communicationtypes'
comtypeid = db.Column(db.Integer, primary_key=True)
comtype = db.Column(db.String(50), unique=True, nullable=False)
description = db.Column(db.Text)
# Types: IP, Serial, USB, VNC, FTP, DNC, Parallel, Network Interface
def __repr__(self):
return f"<CommunicationType {self.comtype}>"
class Communication(BaseModel):
"""
Communication interface for a machine.
Stores network config, serial settings, etc.
"""
__tablename__ = 'communications'
communicationid = db.Column(db.Integer, primary_key=True)
machineid = db.Column(
db.Integer,
db.ForeignKey('machines.machineid'),
nullable=False
)
comtypeid = db.Column(
db.Integer,
db.ForeignKey('communicationtypes.comtypeid'),
nullable=False
)
# Network configuration (for IP type)
ipaddress = db.Column(db.String(50))
subnetmask = db.Column(db.String(50))
gateway = db.Column(db.String(50))
dns1 = db.Column(db.String(50))
dns2 = db.Column(db.String(50))
macaddress = db.Column(db.String(50))
isdhcp = db.Column(db.Boolean, default=False)
# Serial configuration (for Serial type)
comport = db.Column(db.String(20))
baudrate = db.Column(db.Integer)
databits = db.Column(db.Integer)
stopbits = db.Column(db.String(10))
parity = db.Column(db.String(20))
flowcontrol = db.Column(db.String(20))
# VNC/FTP configuration
port = db.Column(db.Integer)
username = db.Column(db.String(100))
# Note: passwords should not be stored here - use secure vault
# DNC configuration
pathname = db.Column(db.String(255))
pathname2 = db.Column(db.String(255), comment='Secondary path for dualpath')
# Flags
isprimary = db.Column(
db.Boolean,
default=False,
comment='Primary communication method'
)
ismachinenetwork = db.Column(
db.Boolean,
default=False,
comment='On machine network vs office network'
)
notes = db.Column(db.Text)
# Relationships
comtype = db.relationship('CommunicationType', backref='communications')
__table_args__ = (
db.Index('idx_comm_machine', 'machineid'),
db.Index('idx_comm_ip', 'ipaddress'),
)
def __repr__(self):
return f"<Communication {self.machineid}:{self.comtype.comtype if self.comtype else 'Unknown'}>"

View File

@@ -0,0 +1,27 @@
"""Knowledge Base models."""
from shopdb.extensions import db
from .base import BaseModel
class KnowledgeBase(BaseModel):
"""Knowledge Base article linking to external resources."""
__tablename__ = 'knowledgebase'
linkid = db.Column(db.Integer, primary_key=True)
appid = db.Column(db.Integer, db.ForeignKey('applications.appid'))
shortdescription = db.Column(db.String(500), nullable=False)
linkurl = db.Column(db.String(2000))
keywords = db.Column(db.String(500))
clicks = db.Column(db.Integer, default=0)
lastupdated = db.Column(db.DateTime, default=db.func.now(), onupdate=db.func.now())
# Relationships
application = db.relationship('Application', backref=db.backref('knowledgebase_articles', lazy='dynamic'))
def __repr__(self):
return f"<KnowledgeBase {self.linkid}: {self.shortdescription[:50] if self.shortdescription else 'No desc'}>"
def increment_clicks(self):
"""Increment click counter."""
self.clicks = (self.clicks or 0) + 1

View File

@@ -0,0 +1,24 @@
"""Location model."""
from shopdb.extensions import db
from .base import BaseModel
class Location(BaseModel):
"""Physical location model."""
__tablename__ = 'locations'
locationid = db.Column(db.Integer, primary_key=True)
locationname = db.Column(db.String(100), unique=True, nullable=False)
building = db.Column(db.String(100))
floor = db.Column(db.String(50))
room = db.Column(db.String(50))
description = db.Column(db.Text)
# Map configuration
mapimage = db.Column(db.String(500), comment='Path to floor map image')
mapwidth = db.Column(db.Integer)
mapheight = db.Column(db.Integer)
def __repr__(self):
return f"<Location {self.locationname}>"

View File

@@ -0,0 +1,252 @@
"""Unified Machine model - combines equipment and PCs."""
from shopdb.extensions import db
from .base import BaseModel, SoftDeleteMixin, AuditMixin
class MachineType(BaseModel):
"""
Machine type classification.
Categories: Equipment, PC, Network, Printer
"""
__tablename__ = 'machinetypes'
machinetypeid = db.Column(db.Integer, primary_key=True)
machinetype = db.Column(db.String(100), unique=True, nullable=False)
category = db.Column(
db.String(50),
nullable=False,
default='Equipment',
comment='Equipment, PC, Network, or Printer'
)
description = db.Column(db.Text)
icon = db.Column(db.String(50), comment='Icon name for UI')
def __repr__(self):
return f"<MachineType {self.machinetype}>"
class MachineStatus(BaseModel):
"""Machine status options."""
__tablename__ = 'machinestatuses'
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"<MachineStatus {self.status}>"
class PCType(BaseModel):
"""
PC type classification for more specific PC categorization.
Examples: Shopfloor PC, Engineer Workstation, CMM PC, etc.
"""
__tablename__ = 'pctypes'
pctypeid = db.Column(db.Integer, primary_key=True)
pctype = db.Column(db.String(100), unique=True, nullable=False)
description = db.Column(db.Text)
def __repr__(self):
return f"<PCType {self.pctype}>"
class Machine(BaseModel, SoftDeleteMixin, AuditMixin):
"""
Unified machine model for all asset types.
Machine types can be:
- CNC machines, CMMs, EDMs, etc. (manufacturing equipment)
- PCs (shopfloor PCs, engineer workstations, etc.)
- Network devices (servers, switches, etc.) - if network_devices plugin not used
The machinetype.category field distinguishes between types.
"""
__tablename__ = 'machines'
machineid = db.Column(db.Integer, primary_key=True)
# Identification
machinenumber = db.Column(
db.String(50),
unique=True,
nullable=False,
index=True,
comment='Business identifier (e.g., CMM01, G5QX1GT3ESF)'
)
alias = db.Column(
db.String(100),
comment='Friendly name'
)
hostname = db.Column(
db.String(100),
index=True,
comment='Network hostname (for PCs)'
)
serialnumber = db.Column(
db.String(100),
index=True,
comment='Hardware serial number'
)
# Classification
machinetypeid = db.Column(
db.Integer,
db.ForeignKey('machinetypes.machinetypeid'),
nullable=False
)
pctypeid = db.Column(
db.Integer,
db.ForeignKey('pctypes.pctypeid'),
nullable=True,
comment='Set for PCs, NULL for equipment'
)
businessunitid = db.Column(
db.Integer,
db.ForeignKey('businessunits.businessunitid'),
nullable=True
)
modelnumberid = db.Column(
db.Integer,
db.ForeignKey('models.modelnumberid'),
nullable=True
)
vendorid = db.Column(
db.Integer,
db.ForeignKey('vendors.vendorid'),
nullable=True
)
# Status
statusid = db.Column(
db.Integer,
db.ForeignKey('machinestatuses.statusid'),
default=1,
comment='In Use, Spare, Retired, etc.'
)
# Location and mapping
locationid = db.Column(
db.Integer,
db.ForeignKey('locations.locationid'),
nullable=True
)
mapleft = db.Column(db.Integer, comment='X coordinate on floor map')
maptop = db.Column(db.Integer, comment='Y coordinate on floor map')
islocationonly = db.Column(
db.Boolean,
default=False,
comment='Virtual location marker (not actual machine)'
)
# PC-specific fields (nullable for non-PC machines)
osid = db.Column(
db.Integer,
db.ForeignKey('operatingsystems.osid'),
nullable=True
)
loggedinuser = db.Column(db.String(100), nullable=True)
lastreporteddate = db.Column(db.DateTime, nullable=True)
lastboottime = db.Column(db.DateTime, nullable=True)
# Features/flags
isvnc = db.Column(db.Boolean, default=False, comment='VNC remote access enabled')
iswinrm = db.Column(db.Boolean, default=False, comment='WinRM enabled')
isshopfloor = db.Column(db.Boolean, default=False, comment='Shopfloor PC')
requiresmanualconfig = db.Column(
db.Boolean,
default=False,
comment='Multi-PC machine needs manual configuration'
)
# Notes
notes = db.Column(db.Text, nullable=True)
# Relationships
machinetype = db.relationship('MachineType', backref='machines')
pctype = db.relationship('PCType', backref='machines')
businessunit = db.relationship('BusinessUnit', backref='machines')
model = db.relationship('Model', backref='machines')
vendor = db.relationship('Vendor', backref='machines')
status = db.relationship('MachineStatus', backref='machines')
location = db.relationship('Location', backref='machines')
operatingsystem = db.relationship('OperatingSystem', backref='machines')
# Communications (one-to-many)
communications = db.relationship(
'Communication',
backref='machine',
cascade='all, delete-orphan',
lazy='dynamic'
)
# Installed applications (for PCs)
installedapps = db.relationship(
'InstalledApp',
back_populates='machine',
cascade='all, delete-orphan',
lazy='dynamic'
)
# Indexes
__table_args__ = (
db.Index('idx_machine_type_bu', 'machinetypeid', 'businessunitid'),
db.Index('idx_machine_location', 'locationid'),
db.Index('idx_machine_active', 'isactive'),
db.Index('idx_machine_hostname', 'hostname'),
)
def __repr__(self):
return f"<Machine {self.machinenumber}>"
@property
def display_name(self):
"""Get display name (alias if set, otherwise machinenumber)."""
return self.alias or self.machinenumber
@property
def derived_machinetype(self):
"""Get machinetype from model (single source of truth)."""
if self.model and self.model.machinetype:
return self.model.machinetype
return None
@property
def is_pc(self):
"""Check if this machine is a PC type."""
mt = self.derived_machinetype
return mt.category == 'PC' if mt else False
@property
def is_equipment(self):
"""Check if this machine is equipment."""
mt = self.derived_machinetype
return mt.category == 'Equipment' if mt else False
@property
def is_network_device(self):
"""Check if this machine is a network device."""
mt = self.derived_machinetype
return mt.category == 'Network' if mt else False
@property
def is_printer(self):
"""Check if this machine is a printer."""
mt = self.derived_machinetype
return mt.category == 'Printer' if mt else False
@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

View File

@@ -0,0 +1,43 @@
"""Model (equipment model number) model."""
from shopdb.extensions import db
from .base import BaseModel
class Model(BaseModel):
"""Equipment/device model information."""
__tablename__ = 'models'
modelnumberid = db.Column(db.Integer, primary_key=True)
modelnumber = db.Column(db.String(100), nullable=False)
# Link to machine type (what kind of equipment this model is for)
machinetypeid = db.Column(
db.Integer,
db.ForeignKey('machinetypes.machinetypeid'),
nullable=True
)
# Link to vendor/manufacturer
vendorid = db.Column(
db.Integer,
db.ForeignKey('vendors.vendorid'),
nullable=True
)
description = db.Column(db.Text)
imageurl = db.Column(db.String(500), comment='URL to product image')
documentationurl = db.Column(db.String(500), comment='URL to documentation')
notes = db.Column(db.Text)
# Relationships
machinetype = db.relationship('MachineType', backref='models')
vendor = db.relationship('Vendor', backref='models')
# Unique constraint on modelnumber + vendor
__table_args__ = (
db.UniqueConstraint('modelnumber', 'vendorid', name='uq_model_vendor'),
)
def __repr__(self):
return f"<Model {self.modelnumber}>"

View File

@@ -0,0 +1,22 @@
"""Operating System model."""
from shopdb.extensions import db
from .base import BaseModel
class OperatingSystem(BaseModel):
"""Operating system model."""
__tablename__ = 'operatingsystems'
osid = db.Column(db.Integer, primary_key=True)
osname = db.Column(db.String(100), nullable=False)
osversion = db.Column(db.String(50))
architecture = db.Column(db.String(20), comment='x86, x64, ARM')
endoflife = db.Column(db.Date, comment='End of support date')
__table_args__ = (
db.UniqueConstraint('osname', 'osversion', name='uq_os_name_version'),
)
def __repr__(self):
return f"<OperatingSystem {self.osname} {self.osversion}>"

View File

@@ -0,0 +1,77 @@
"""Machine relationship models."""
from shopdb.extensions import db
from .base import BaseModel
class RelationshipType(BaseModel):
"""Types of relationships between machines."""
__tablename__ = 'relationshiptypes'
relationshiptypeid = db.Column(db.Integer, primary_key=True)
relationshiptype = db.Column(db.String(50), unique=True, nullable=False)
description = db.Column(db.Text)
# Example types:
# - "Controls" (PC controls Equipment)
# - "Dualpath" (Redundant path partner)
# - "Backup" (Backup machine)
def __repr__(self):
return f"<RelationshipType {self.relationshiptype}>"
class MachineRelationship(BaseModel):
"""
Relationships between machines.
Examples:
- PC controls CNC machine
- Two CNCs are dualpath partners
"""
__tablename__ = 'machinerelationships'
relationshipid = db.Column(db.Integer, primary_key=True)
parentmachineid = db.Column(
db.Integer,
db.ForeignKey('machines.machineid'),
nullable=False
)
childmachineid = db.Column(
db.Integer,
db.ForeignKey('machines.machineid'),
nullable=False
)
relationshiptypeid = db.Column(
db.Integer,
db.ForeignKey('relationshiptypes.relationshiptypeid'),
nullable=False
)
notes = db.Column(db.Text)
# Relationships
parent_machine = db.relationship(
'Machine',
foreign_keys=[parentmachineid],
backref='child_relationships'
)
child_machine = db.relationship(
'Machine',
foreign_keys=[childmachineid],
backref='parent_relationships'
)
relationship_type = db.relationship('RelationshipType', backref='relationships')
__table_args__ = (
db.UniqueConstraint(
'parentmachineid',
'childmachineid',
'relationshiptypeid',
name='uq_machine_relationship'
),
)
def __repr__(self):
return f"<MachineRelationship {self.parentmachineid} -> {self.childmachineid}>"

View File

@@ -0,0 +1,73 @@
"""User and authentication models."""
from datetime import datetime
from shopdb.extensions import db
from .base import BaseModel
# Association table for user roles (many-to-many)
userroles = db.Table(
'userroles',
db.Column('userid', db.Integer, db.ForeignKey('users.userid'), primary_key=True),
db.Column('roleid', db.Integer, db.ForeignKey('roles.roleid'), primary_key=True)
)
class Role(BaseModel):
"""User role model."""
__tablename__ = 'roles'
roleid = db.Column(db.Integer, primary_key=True)
rolename = db.Column(db.String(50), unique=True, nullable=False)
description = db.Column(db.Text)
def __repr__(self):
return f"<Role {self.rolename}>"
class User(BaseModel):
"""User model for authentication."""
__tablename__ = 'users'
userid = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(100), unique=True, nullable=False, index=True)
email = db.Column(db.String(255), unique=True, nullable=False)
passwordhash = db.Column(db.String(255), nullable=False)
# Profile
firstname = db.Column(db.String(100))
lastname = db.Column(db.String(100))
# Status
lastlogindate = db.Column(db.DateTime)
failedlogins = db.Column(db.Integer, default=0)
lockeduntil = db.Column(db.DateTime)
# Relationships
roles = db.relationship(
'Role',
secondary=userroles,
backref=db.backref('users', lazy='dynamic')
)
def __repr__(self):
return f"<User {self.username}>"
@property
def islocked(self):
"""Check if account is locked."""
if self.lockeduntil:
return datetime.utcnow() < self.lockeduntil
return False
def hasrole(self, rolename: str) -> bool:
"""Check if user has a specific role."""
return any(r.rolename == rolename for r in self.roles)
def getpermissions(self) -> list:
"""Get list of permission names from roles."""
# Simple role-based permissions
perms = []
for role in self.roles:
perms.append(role.rolename)
return perms

View File

@@ -0,0 +1,20 @@
"""Vendor model."""
from shopdb.extensions import db
from .base import BaseModel
class Vendor(BaseModel):
"""Vendor/Manufacturer model."""
__tablename__ = 'vendors'
vendorid = db.Column(db.Integer, primary_key=True)
vendor = db.Column(db.String(100), unique=True, nullable=False)
description = db.Column(db.Text)
website = db.Column(db.String(255))
supportphone = db.Column(db.String(50))
supportemail = db.Column(db.String(100))
notes = db.Column(db.Text)
def __repr__(self):
return f"<Vendor {self.vendor}>"

View File

@@ -0,0 +1 @@
"""Core Marshmallow schemas."""

View File

@@ -0,0 +1 @@
"""Core services."""

54
shopdb/exceptions.py Normal file
View File

@@ -0,0 +1,54 @@
"""Custom exceptions for ShopDB."""
class ShopDBException(Exception):
"""Base exception for ShopDB."""
def __init__(self, message: str, code: str = None, details: dict = None):
super().__init__(message)
self.message = message
self.code = code or 'SHOPDB_ERROR'
self.details = details or {}
class ValidationError(ShopDBException):
"""Validation error."""
def __init__(self, message: str, details: dict = None):
super().__init__(message, 'VALIDATION_ERROR', details)
class NotFoundError(ShopDBException):
"""Resource not found error."""
def __init__(self, resource: str, identifier):
message = f"{resource} with ID {identifier} not found"
super().__init__(message, 'NOT_FOUND', {'resource': resource, 'id': identifier})
class AuthenticationError(ShopDBException):
"""Authentication error."""
def __init__(self, message: str = "Authentication required"):
super().__init__(message, 'UNAUTHORIZED')
class AuthorizationError(ShopDBException):
"""Authorization error."""
def __init__(self, message: str = "Permission denied"):
super().__init__(message, 'FORBIDDEN')
class ConflictError(ShopDBException):
"""Conflict error (e.g., duplicate entry)."""
def __init__(self, message: str, details: dict = None):
super().__init__(message, 'CONFLICT', details)
class PluginError(ShopDBException):
"""Plugin-related error."""
def __init__(self, message: str, plugin_name: str = None):
super().__init__(message, 'PLUGIN_ERROR', {'plugin': plugin_name})

25
shopdb/extensions.py Normal file
View File

@@ -0,0 +1,25 @@
"""Flask extensions initialization."""
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_jwt_extended import JWTManager
from flask_cors import CORS
from flask_marshmallow import Marshmallow
# Initialize extensions without app
db = SQLAlchemy()
migrate = Migrate()
jwt = JWTManager()
cors = CORS()
ma = Marshmallow()
def init_extensions(app):
"""Initialize all Flask extensions with app."""
db.init_app(app)
migrate.init_app(app, db)
jwt.init_app(app)
cors.init_app(app, resources={
r"/api/*": {"origins": app.config.get('CORS_ORIGINS', '*')}
})
ma.init_app(app)

276
shopdb/plugins/__init__.py Normal file
View File

@@ -0,0 +1,276 @@
"""Plugin manager - main entry point for plugin system."""
from pathlib import Path
from typing import Dict, List, Optional
from flask import Flask
import logging
from .base import BasePlugin, PluginMeta
from .registry import PluginRegistry, PluginState
from .loader import PluginLoader
from .migrations import PluginMigrationManager
logger = logging.getLogger(__name__)
__all__ = [
'PluginManager',
'BasePlugin',
'PluginMeta',
'PluginRegistry',
'PluginState',
'plugin_manager'
]
class PluginManager:
"""
Central manager for all plugin operations.
Usage:
plugin_manager = PluginManager()
plugin_manager.init_app(app, db)
# In CLI:
plugin_manager.install_plugin('printers')
"""
def __init__(self):
self.registry: Optional[PluginRegistry] = None
self.loader: Optional[PluginLoader] = None
self.migration_manager: Optional[PluginMigrationManager] = None
self._app: Optional[Flask] = None
self._db = None
def init_app(self, app: Flask, db) -> None:
"""Initialize plugin manager with Flask app."""
self._app = app
self._db = db
# Setup paths
instance_path = Path(app.instance_path)
plugins_dir = Path(app.root_path).parent / 'plugins'
# Initialize components
self.registry = PluginRegistry(instance_path / 'plugins.json')
self.loader = PluginLoader(plugins_dir, self.registry)
self.migration_manager = PluginMigrationManager(
plugins_dir,
app.config.get('SQLALCHEMY_DATABASE_URI')
)
# Load enabled plugins
self._load_enabled_plugins()
# Store on app for access
app.extensions['plugin_manager'] = self
def _load_enabled_plugins(self) -> None:
"""Load and register all enabled plugins."""
plugins = self.loader.load_enabled_plugins(self._app, self._db)
for name, plugin in plugins.items():
self._register_plugin_components(plugin)
def _register_plugin_components(self, plugin: BasePlugin) -> None:
"""Register plugin's blueprint, models, CLI commands, etc."""
# Register blueprint
blueprint = plugin.get_blueprint()
if blueprint:
self._app.register_blueprint(
blueprint,
url_prefix=plugin.meta.api_prefix
)
logger.debug(f"Registered blueprint: {plugin.meta.api_prefix}")
# Register CLI commands
for cmd in plugin.get_cli_commands():
self._app.cli.add_command(cmd)
def discover_available(self) -> List[Dict]:
"""
Get list of all available plugins (installed or not).
Returns list of plugin info dicts.
"""
available = []
for name in self.loader.discover_plugins():
plugin_class = self.loader.load_plugin_class(name)
if plugin_class:
try:
temp = plugin_class()
meta = temp.meta
state = self.registry.get(name)
available.append({
'name': meta.name,
'version': meta.version,
'description': meta.description,
'author': meta.author,
'dependencies': meta.dependencies,
'installed': state is not None,
'enabled': state.enabled if state else False,
'installed_at': state.installed_at if state else None
})
except Exception as e:
logger.warning(f"Error inspecting plugin {name}: {e}")
return available
def install_plugin(self, name: str, run_migrations: bool = True) -> bool:
"""
Install a plugin.
Steps:
1. Verify plugin exists
2. Check dependencies
3. Run database migrations
4. Register in registry
5. Call plugin's on_install hook
"""
# Check if already installed
if self.registry.is_installed(name):
logger.warning(f"Plugin {name} is already installed")
return False
# Load plugin class
plugin_class = self.loader.load_plugin_class(name)
if not plugin_class:
logger.error(f"Plugin {name} not found")
return False
temp_plugin = plugin_class()
meta = temp_plugin.meta
# Check dependencies
for dep in meta.dependencies:
if not self.registry.is_installed(dep):
logger.error(
f"Plugin {name} requires {dep} to be installed first"
)
return False
# Run migrations
if run_migrations:
success = self.migration_manager.run_plugin_migrations(name)
if not success:
logger.error(f"Failed to run migrations for {name}")
return False
# Register plugin
self.registry.register(name, meta.version)
# Load the plugin
plugin = self.loader.load_plugin(name, self._app, self._db)
if plugin:
self._register_plugin_components(plugin)
plugin.on_install(self._app)
logger.info(f"Installed plugin: {name} v{meta.version}")
return True
def uninstall_plugin(self, name: str, remove_data: bool = False) -> bool:
"""
Uninstall a plugin.
Args:
name: Plugin name
remove_data: If True, run downgrade migrations to remove tables
"""
if not self.registry.is_installed(name):
logger.warning(f"Plugin {name} is not installed")
return False
# Check if other plugins depend on this one
for other_name in self.registry.get_enabled_plugins():
if other_name == name:
continue
other_plugin = self.loader.get_loaded_plugin(other_name)
if other_plugin and name in other_plugin.meta.dependencies:
logger.error(
f"Cannot uninstall {name}: {other_name} depends on it"
)
return False
# Get plugin instance
plugin = self.loader.get_loaded_plugin(name)
# Call on_uninstall hook
if plugin:
plugin.on_uninstall(self._app)
# Optionally remove data
if remove_data:
self.migration_manager.downgrade_plugin(name)
# Unregister
self.registry.unregister(name)
logger.info(f"Uninstalled plugin: {name}")
return True
def enable_plugin(self, name: str) -> bool:
"""Enable a disabled plugin."""
if not self.registry.is_installed(name):
logger.error(f"Plugin {name} is not installed")
return False
if self.registry.is_enabled(name):
logger.info(f"Plugin {name} is already enabled")
return True
# Check dependencies are enabled
plugin_class = self.loader.load_plugin_class(name)
if plugin_class:
temp = plugin_class()
for dep in temp.meta.dependencies:
if not self.registry.is_enabled(dep):
logger.error(f"Cannot enable {name}: {dep} is not enabled")
return False
self.registry.enable(name)
# Load the plugin
plugin = self.loader.load_plugin(name, self._app, self._db)
if plugin:
self._register_plugin_components(plugin)
plugin.on_enable(self._app)
logger.info(f"Enabled plugin: {name}")
return True
def disable_plugin(self, name: str) -> bool:
"""Disable an enabled plugin."""
if not self.registry.is_enabled(name):
logger.info(f"Plugin {name} is already disabled")
return True
# Check if other plugins depend on this one
for other_name in self.registry.get_enabled_plugins():
if other_name == name:
continue
other_plugin = self.loader.get_loaded_plugin(other_name)
if other_plugin and name in other_plugin.meta.dependencies:
logger.error(
f"Cannot disable {name}: {other_name} depends on it"
)
return False
plugin = self.loader.get_loaded_plugin(name)
if plugin:
plugin.on_disable(self._app)
self.registry.disable(name)
logger.info(f"Disabled plugin: {name}")
return True
def get_plugin(self, name: str) -> Optional[BasePlugin]:
"""Get a loaded plugin instance."""
return self.loader.get_loaded_plugin(name)
def get_all_plugins(self) -> Dict[str, BasePlugin]:
"""Get all loaded plugins."""
return self.loader.get_all_loaded()
# Global plugin manager instance
plugin_manager = PluginManager()

122
shopdb/plugins/base.py Normal file
View File

@@ -0,0 +1,122 @@
"""Base plugin class that all plugins must inherit from."""
from abc import ABC, abstractmethod
from typing import List, Dict, Optional, Type
from dataclasses import dataclass, field
from flask import Flask, Blueprint
@dataclass
class PluginMeta:
"""Plugin metadata container."""
name: str
version: str
description: str
author: str = ""
dependencies: List[str] = field(default_factory=list)
core_version: str = ">=1.0.0"
api_prefix: str = None
def __post_init__(self):
if self.api_prefix is None:
self.api_prefix = f"/api/{self.name.replace('_', '-')}"
class BasePlugin(ABC):
"""
Base class for all ShopDB plugins.
Plugins must implement:
- meta: PluginMeta instance
- get_blueprint(): Return Flask Blueprint for API routes
- get_models(): Return list of SQLAlchemy model classes
Optionally implement:
- init_app(app, db): Custom initialization
- get_cli_commands(): Return Click commands
- get_services(): Return service classes
- on_install(): Called when plugin is installed
- on_uninstall(): Called when plugin is uninstalled
- on_enable(): Called when plugin is enabled
- on_disable(): Called when plugin is disabled
"""
@property
@abstractmethod
def meta(self) -> PluginMeta:
"""Return plugin metadata."""
pass
@abstractmethod
def get_blueprint(self) -> Optional[Blueprint]:
"""Return Flask Blueprint with API routes."""
pass
@abstractmethod
def get_models(self) -> List[Type]:
"""Return list of SQLAlchemy model classes."""
pass
def init_app(self, app: Flask, db) -> None:
"""
Initialize plugin with Flask app.
Override for custom initialization.
"""
pass
def get_cli_commands(self) -> List:
"""Return list of Click command groups/commands."""
return []
def get_services(self) -> Dict[str, Type]:
"""Return dict of service name -> service class."""
return {}
def get_event_handlers(self) -> Dict[str, callable]:
"""Return dict of event name -> handler function."""
return {}
def on_install(self, app: Flask) -> None:
"""Called when plugin is installed via CLI."""
pass
def on_uninstall(self, app: Flask) -> None:
"""Called when plugin is uninstalled via CLI."""
pass
def on_enable(self, app: Flask) -> None:
"""Called when plugin is enabled."""
pass
def on_disable(self, app: Flask) -> None:
"""Called when plugin is disabled."""
pass
def get_dashboard_widgets(self) -> List[Dict]:
"""
Return dashboard widget definitions.
Each widget: {
'name': str,
'component': str, # Frontend component name
'endpoint': str, # API endpoint for data
'size': str, # 'small', 'medium', 'large'
'position': int # Order on dashboard
}
"""
return []
def get_navigation_items(self) -> List[Dict]:
"""
Return navigation menu items.
Each item: {
'name': str,
'icon': str,
'route': str,
'position': int,
'children': []
}
"""
return []

203
shopdb/plugins/cli.py Normal file
View File

@@ -0,0 +1,203 @@
"""Flask CLI commands for plugin management."""
import click
from flask import current_app
from flask.cli import with_appcontext
@click.group('plugin')
def plugin_cli():
"""Plugin management commands."""
pass
@plugin_cli.command('list')
@with_appcontext
def list_plugins():
"""List all available plugins."""
pm = current_app.extensions.get('plugin_manager')
if not pm:
click.echo(click.style("Plugin manager not initialized", fg='red'))
return
plugins = pm.discover_available()
if not plugins:
click.echo("No plugins found in plugins directory.")
return
# Format output
click.echo("")
click.echo(click.style("Available Plugins:", fg='cyan', bold=True))
click.echo("-" * 60)
for p in plugins:
if p['enabled']:
status = click.style("[Enabled]", fg='green')
elif p['installed']:
status = click.style("[Disabled]", fg='yellow')
else:
status = click.style("[Available]", fg='white')
click.echo(f" {p['name']:20} v{p['version']:10} {status}")
if p['description']:
click.echo(f" {p['description'][:55]}...")
if p['dependencies']:
deps = ', '.join(p['dependencies'])
click.echo(f" Dependencies: {deps}")
click.echo("")
@plugin_cli.command('install')
@click.argument('name')
@click.option('--skip-migrations', is_flag=True, help='Skip database migrations')
@with_appcontext
def install_plugin(name: str, skip_migrations: bool):
"""
Install a plugin.
Usage: flask plugin install printers
"""
pm = current_app.extensions.get('plugin_manager')
if not pm:
click.echo(click.style("Plugin manager not initialized", fg='red'))
raise SystemExit(1)
click.echo(f"Installing plugin: {name}")
if pm.install_plugin(name, run_migrations=not skip_migrations):
click.echo(click.style(f"Successfully installed {name}", fg='green'))
else:
click.echo(click.style(f"Failed to install {name}", fg='red'))
raise SystemExit(1)
@plugin_cli.command('uninstall')
@click.argument('name')
@click.option('--remove-data', is_flag=True, help='Remove plugin database tables')
@click.confirmation_option(prompt='Are you sure you want to uninstall this plugin?')
@with_appcontext
def uninstall_plugin(name: str, remove_data: bool):
"""
Uninstall a plugin.
Usage: flask plugin uninstall printers
"""
pm = current_app.extensions.get('plugin_manager')
if not pm:
click.echo(click.style("Plugin manager not initialized", fg='red'))
raise SystemExit(1)
click.echo(f"Uninstalling plugin: {name}")
if pm.uninstall_plugin(name, remove_data=remove_data):
click.echo(click.style(f"Successfully uninstalled {name}", fg='green'))
else:
click.echo(click.style(f"Failed to uninstall {name}", fg='red'))
raise SystemExit(1)
@plugin_cli.command('enable')
@click.argument('name')
@with_appcontext
def enable_plugin(name: str):
"""Enable a disabled plugin."""
pm = current_app.extensions.get('plugin_manager')
if not pm:
click.echo(click.style("Plugin manager not initialized", fg='red'))
raise SystemExit(1)
if pm.enable_plugin(name):
click.echo(click.style(f"Enabled {name}", fg='green'))
else:
click.echo(click.style(f"Failed to enable {name}", fg='red'))
raise SystemExit(1)
@plugin_cli.command('disable')
@click.argument('name')
@with_appcontext
def disable_plugin(name: str):
"""Disable an enabled plugin."""
pm = current_app.extensions.get('plugin_manager')
if not pm:
click.echo(click.style("Plugin manager not initialized", fg='red'))
raise SystemExit(1)
if pm.disable_plugin(name):
click.echo(click.style(f"Disabled {name}", fg='green'))
else:
click.echo(click.style(f"Failed to disable {name}", fg='red'))
raise SystemExit(1)
@plugin_cli.command('info')
@click.argument('name')
@with_appcontext
def plugin_info(name: str):
"""Show detailed information about a plugin."""
pm = current_app.extensions.get('plugin_manager')
if not pm:
click.echo(click.style("Plugin manager not initialized", fg='red'))
raise SystemExit(1)
plugin_class = pm.loader.load_plugin_class(name)
if not plugin_class:
click.echo(click.style(f"Plugin {name} not found", fg='red'))
raise SystemExit(1)
try:
temp = plugin_class()
meta = temp.meta
except Exception as e:
click.echo(click.style(f"Error loading plugin: {e}", fg='red'))
raise SystemExit(1)
state = pm.registry.get(name)
click.echo("")
click.echo("=" * 50)
click.echo(click.style(f"Plugin: {meta.name}", fg='cyan', bold=True))
click.echo("=" * 50)
click.echo(f"Version: {meta.version}")
click.echo(f"Description: {meta.description}")
click.echo(f"Author: {meta.author or 'Unknown'}")
click.echo(f"API Prefix: {meta.api_prefix}")
click.echo(f"Dependencies: {', '.join(meta.dependencies) or 'None'}")
click.echo(f"Core Version: {meta.core_version}")
click.echo("")
if state:
status = click.style('Enabled', fg='green') if state.enabled else click.style('Disabled', fg='yellow')
click.echo(f"Status: {status}")
click.echo(f"Installed: {state.installed_at}")
click.echo(f"Migrations: {len(state.migrations_applied)} applied")
else:
click.echo(f"Status: {click.style('Not installed', fg='white')}")
click.echo("")
@plugin_cli.command('migrate')
@click.argument('name')
@click.option('--revision', default='head', help='Target revision')
@with_appcontext
def migrate_plugin(name: str, revision: str):
"""Run migrations for a specific plugin."""
pm = current_app.extensions.get('plugin_manager')
if not pm:
click.echo(click.style("Plugin manager not initialized", fg='red'))
raise SystemExit(1)
if not pm.registry.is_installed(name):
click.echo(click.style(f"Plugin {name} is not installed", fg='red'))
raise SystemExit(1)
click.echo(f"Running migrations for {name}...")
if pm.migration_manager.run_plugin_migrations(name, revision):
click.echo(click.style("Migrations completed", fg='green'))
else:
click.echo(click.style("Migration failed", fg='red'))
raise SystemExit(1)

174
shopdb/plugins/loader.py Normal file
View File

@@ -0,0 +1,174 @@
"""Plugin discovery and loading."""
import importlib
import importlib.util
from pathlib import Path
from typing import Dict, List, Type, Optional
from flask import Flask
import logging
from .base import BasePlugin
from .registry import PluginRegistry
logger = logging.getLogger(__name__)
class PluginLoader:
"""
Discovers and loads plugins from the plugins directory.
"""
def __init__(self, plugins_dir: Path, registry: PluginRegistry):
self.plugins_dir = plugins_dir
self.registry = registry
self._loaded_plugins: Dict[str, BasePlugin] = {}
self._plugin_classes: Dict[str, Type[BasePlugin]] = {}
def discover_plugins(self) -> List[str]:
"""
Discover available plugins in plugins directory.
Returns list of plugin names.
"""
available = []
if not self.plugins_dir.exists():
return available
for item in self.plugins_dir.iterdir():
if item.is_dir() and (item / 'plugin.py').exists():
available.append(item.name)
elif item.is_dir() and (item / '__init__.py').exists():
# Check for plugin.py in package
if (item / 'plugin.py').exists():
available.append(item.name)
return available
def load_plugin_class(self, name: str) -> Optional[Type[BasePlugin]]:
"""
Load plugin class without instantiating.
Used for inspection before installation.
"""
if name in self._plugin_classes:
return self._plugin_classes[name]
plugin_dir = self.plugins_dir / name
plugin_module_path = plugin_dir / 'plugin.py'
if not plugin_module_path.exists():
logger.error(f"Plugin {name} not found: {plugin_module_path}")
return None
try:
# Import the plugin module
spec = importlib.util.spec_from_file_location(
f"plugins.{name}.plugin",
plugin_module_path
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
# Find the plugin class
for attr_name in dir(module):
attr = getattr(module, attr_name)
if (isinstance(attr, type) and
issubclass(attr, BasePlugin) and
attr is not BasePlugin):
self._plugin_classes[name] = attr
return attr
logger.error(f"No BasePlugin subclass found in {name}")
return None
except Exception as e:
logger.error(f"Error loading plugin {name}: {e}")
return None
def load_plugin(self, name: str, app: Flask, db) -> Optional[BasePlugin]:
"""
Load and instantiate a plugin.
"""
if name in self._loaded_plugins:
return self._loaded_plugins[name]
plugin_class = self.load_plugin_class(name)
if not plugin_class:
return None
try:
# Instantiate plugin
plugin = plugin_class()
# Check dependencies
for dep in plugin.meta.dependencies:
if not self.registry.is_enabled(dep):
logger.error(
f"Plugin {name} requires {dep} which is not enabled"
)
return None
# Initialize plugin
plugin.init_app(app, db)
self._loaded_plugins[name] = plugin
return plugin
except Exception as e:
logger.error(f"Error instantiating plugin {name}: {e}")
return None
def load_enabled_plugins(self, app: Flask, db) -> Dict[str, BasePlugin]:
"""
Load all enabled plugins.
Returns dict of name -> plugin instance.
"""
loaded = {}
# Sort by dependencies (simple topological sort)
enabled = self.registry.get_enabled_plugins()
sorted_plugins = self._sort_by_dependencies(enabled)
for name in sorted_plugins:
plugin = self.load_plugin(name, app, db)
if plugin:
loaded[name] = plugin
logger.info(f"Loaded plugin: {name} v{plugin.meta.version}")
else:
logger.warning(f"Failed to load plugin: {name}")
return loaded
def _sort_by_dependencies(self, plugin_names: List[str]) -> List[str]:
"""Sort plugins so dependencies come first."""
sorted_list = []
visited = set()
def visit(name):
if name in visited:
return
visited.add(name)
plugin_class = self.load_plugin_class(name)
if plugin_class:
# Create temporary instance to get meta
try:
temp = plugin_class()
for dep in temp.meta.dependencies:
if dep in plugin_names:
visit(dep)
except Exception:
pass
sorted_list.append(name)
for name in plugin_names:
visit(name)
return sorted_list
def get_loaded_plugin(self, name: str) -> Optional[BasePlugin]:
"""Get an already loaded plugin."""
return self._loaded_plugins.get(name)
def get_all_loaded(self) -> Dict[str, BasePlugin]:
"""Get all loaded plugins."""
return self._loaded_plugins.copy()

View File

@@ -0,0 +1,173 @@
"""Plugin migration management using Alembic."""
from pathlib import Path
from typing import Optional
import logging
import subprocess
import sys
logger = logging.getLogger(__name__)
class PluginMigrationManager:
"""
Manages database migrations for plugins.
Each plugin has its own migrations directory.
"""
def __init__(self, plugins_dir: Path, database_url: str):
self.plugins_dir = plugins_dir
self.database_url = database_url
def get_migrations_dir(self, plugin_name: str) -> Optional[Path]:
"""Get migrations directory for a plugin."""
migrations_dir = self.plugins_dir / plugin_name / 'migrations'
if migrations_dir.exists():
return migrations_dir
return None
def run_plugin_migrations(
self,
plugin_name: str,
revision: str = 'head'
) -> bool:
"""
Run migrations for a plugin.
Uses flask db upgrade with the plugin's migrations directory.
"""
migrations_dir = self.get_migrations_dir(plugin_name)
if not migrations_dir:
logger.info(f"No migrations directory for plugin {plugin_name}")
return True # No migrations to run
try:
# Use alembic directly with plugin's migrations
from alembic.config import Config
from alembic import command
config = Config()
config.set_main_option('script_location', str(migrations_dir))
config.set_main_option('sqlalchemy.url', self.database_url)
# Use plugin-specific version table
config.set_main_option(
'version_table',
f'alembic_version_{plugin_name}'
)
command.upgrade(config, revision)
logger.info(f"Migrations completed for {plugin_name}")
return True
except ImportError:
# Fallback to subprocess if alembic not available in context
logger.warning("Using subprocess for migrations")
return self._run_migrations_subprocess(plugin_name, revision)
except Exception as e:
logger.error(f"Migration failed for {plugin_name}: {e}")
return False
def _run_migrations_subprocess(
self,
plugin_name: str,
revision: str = 'head'
) -> bool:
"""Run migrations via subprocess as fallback."""
migrations_dir = self.get_migrations_dir(plugin_name)
if not migrations_dir:
return True
try:
result = subprocess.run(
[
sys.executable, '-m', 'alembic',
'-c', str(migrations_dir / 'alembic.ini'),
'upgrade', revision
],
capture_output=True,
text=True,
env={
**dict(__import__('os').environ),
'DATABASE_URL': self.database_url
}
)
if result.returncode != 0:
logger.error(f"Migration error: {result.stderr}")
return False
return True
except Exception as e:
logger.error(f"Migration subprocess failed: {e}")
return False
def downgrade_plugin(
self,
plugin_name: str,
revision: str = 'base'
) -> bool:
"""
Downgrade/rollback plugin migrations.
"""
migrations_dir = self.get_migrations_dir(plugin_name)
if not migrations_dir:
return True
try:
from alembic.config import Config
from alembic import command
config = Config()
config.set_main_option('script_location', str(migrations_dir))
config.set_main_option('sqlalchemy.url', self.database_url)
config.set_main_option(
'version_table',
f'alembic_version_{plugin_name}'
)
command.downgrade(config, revision)
logger.info(f"Downgrade completed for {plugin_name}")
return True
except Exception as e:
logger.error(f"Downgrade failed for {plugin_name}: {e}")
return False
def get_current_revision(self, plugin_name: str) -> Optional[str]:
"""Get current migration revision for a plugin."""
migrations_dir = self.get_migrations_dir(plugin_name)
if not migrations_dir:
return None
try:
from alembic.config import Config
from alembic.script import ScriptDirectory
config = Config()
config.set_main_option('script_location', str(migrations_dir))
script = ScriptDirectory.from_config(config)
return script.get_current_head()
except Exception:
return None
def has_pending_migrations(self, plugin_name: str) -> bool:
"""Check if plugin has pending migrations."""
# Simplified check - would need DB connection for full check
migrations_dir = self.get_migrations_dir(plugin_name)
if not migrations_dir:
return False
versions_dir = migrations_dir / 'versions'
if not versions_dir.exists():
return False
# Check if there are any migration files
migration_files = list(versions_dir.glob('*.py'))
return len(migration_files) > 0

121
shopdb/plugins/registry.py Normal file
View File

@@ -0,0 +1,121 @@
"""Plugin registry for tracking installed and enabled plugins."""
import json
from pathlib import Path
from typing import Dict, List, Optional
from dataclasses import dataclass, field, asdict
from datetime import datetime
@dataclass
class PluginState:
"""Persistent state for a plugin."""
name: str
version: str
installed_at: str
enabled: bool = True
migrations_applied: List[str] = field(default_factory=list)
config: Dict = field(default_factory=dict)
class PluginRegistry:
"""
Manages plugin state persistence.
Stores state in JSON file in instance folder.
"""
def __init__(self, state_file: Path):
self.state_file = state_file
self._plugins: Dict[str, PluginState] = {}
self._load()
def _load(self) -> None:
"""Load registry from file."""
if self.state_file.exists():
try:
with open(self.state_file, 'r') as f:
data = json.load(f)
for name, state_data in data.get('plugins', {}).items():
self._plugins[name] = PluginState(**state_data)
except (json.JSONDecodeError, TypeError):
# Corrupted file, start fresh
self._plugins = {}
def _save(self) -> None:
"""Save registry to file."""
self.state_file.parent.mkdir(parents=True, exist_ok=True)
with open(self.state_file, 'w') as f:
json.dump({
'plugins': {
name: asdict(state)
for name, state in self._plugins.items()
}
}, f, indent=2)
def register(self, name: str, version: str) -> PluginState:
"""Register a newly installed plugin."""
state = PluginState(
name=name,
version=version,
installed_at=datetime.utcnow().isoformat(),
enabled=True
)
self._plugins[name] = state
self._save()
return state
def unregister(self, name: str) -> None:
"""Remove plugin from registry."""
if name in self._plugins:
del self._plugins[name]
self._save()
def get(self, name: str) -> Optional[PluginState]:
"""Get plugin state."""
return self._plugins.get(name)
def is_installed(self, name: str) -> bool:
"""Check if plugin is installed."""
return name in self._plugins
def is_enabled(self, name: str) -> bool:
"""Check if plugin is enabled."""
state = self._plugins.get(name)
return state.enabled if state else False
def enable(self, name: str) -> None:
"""Enable a plugin."""
if name in self._plugins:
self._plugins[name].enabled = True
self._save()
def disable(self, name: str) -> None:
"""Disable a plugin."""
if name in self._plugins:
self._plugins[name].enabled = False
self._save()
def get_enabled_plugins(self) -> List[str]:
"""Get list of enabled plugin names."""
return [
name for name, state in self._plugins.items()
if state.enabled
]
def add_migration(self, name: str, revision: str) -> None:
"""Record that a migration was applied."""
if name in self._plugins:
if revision not in self._plugins[name].migrations_applied:
self._plugins[name].migrations_applied.append(revision)
self._save()
def get_all(self) -> Dict[str, PluginState]:
"""Get all registered plugins."""
return self._plugins.copy()
def update_config(self, name: str, config: Dict) -> None:
"""Update plugin configuration."""
if name in self._plugins:
self._plugins[name].config.update(config)
self._save()

Binary file not shown.

After

Width:  |  Height:  |  Size: 832 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 828 KiB

20
shopdb/utils/__init__.py Normal file
View File

@@ -0,0 +1,20 @@
"""Utility modules."""
from .responses import (
api_response,
success_response,
error_response,
paginated_response,
ErrorCodes
)
from .pagination import get_pagination_params, paginate_query
__all__ = [
'api_response',
'success_response',
'error_response',
'paginated_response',
'ErrorCodes',
'get_pagination_params',
'paginate_query'
]

View File

@@ -0,0 +1,43 @@
"""Pagination utilities."""
from flask import request, current_app
from typing import Tuple
def get_pagination_params(req=None) -> Tuple[int, int]:
"""
Extract pagination parameters from request.
Returns:
Tuple of (page, per_page)
"""
if req is None:
req = request
default_size = current_app.config.get('DEFAULT_PAGE_SIZE', 20)
max_size = current_app.config.get('MAX_PAGE_SIZE', 100)
try:
page = max(1, int(req.args.get('page', 1)))
except (TypeError, ValueError):
page = 1
try:
per_page = int(req.args.get('per_page', default_size))
per_page = max(1, min(per_page, max_size))
except (TypeError, ValueError):
per_page = default_size
return page, per_page
def paginate_query(query, page: int, per_page: int):
"""
Apply pagination to a SQLAlchemy query.
Returns:
Tuple of (items, total)
"""
total = query.count()
items = query.offset((page - 1) * per_page).limit(per_page).all()
return items, total

171
shopdb/utils/responses.py Normal file
View File

@@ -0,0 +1,171 @@
"""Standardized API response helpers."""
from flask import jsonify, make_response
from typing import Any, Dict, List, Optional
from datetime import datetime
import uuid
class ErrorCodes:
"""Standard error codes."""
VALIDATION_ERROR = 'VALIDATION_ERROR'
NOT_FOUND = 'NOT_FOUND'
UNAUTHORIZED = 'UNAUTHORIZED'
FORBIDDEN = 'FORBIDDEN'
CONFLICT = 'CONFLICT'
INTERNAL_ERROR = 'INTERNAL_ERROR'
BAD_REQUEST = 'BAD_REQUEST'
PLUGIN_ERROR = 'PLUGIN_ERROR'
def api_response(
data: Any = None,
message: str = None,
status: str = 'success',
meta: Dict = None,
http_code: int = 200
):
"""
Create standardized API response.
Response format:
{
"status": "success" | "error",
"data": {...} | [...],
"message": "Optional message",
"meta": {
"timestamp": "2025-01-12T...",
"request_id": "uuid"
}
}
"""
response = {
'status': status,
'meta': {
'timestamp': datetime.utcnow().isoformat() + 'Z',
'request_id': str(uuid.uuid4())[:8],
**(meta or {})
}
}
if data is not None:
response['data'] = data
if message:
response['message'] = message
return make_response(jsonify(response), http_code)
def success_response(
data: Any = None,
message: str = None,
meta: Dict = None,
http_code: int = 200
):
"""Success response helper."""
return api_response(
data=data,
message=message,
meta=meta,
status='success',
http_code=http_code
)
def error_response(
code: str,
message: str,
details: Dict = None,
http_code: int = 400
):
"""
Error response helper.
Response format:
{
"status": "error",
"error": {
"code": "VALIDATION_ERROR",
"message": "Human-readable message",
"details": {...}
}
}
"""
error_data = {
'code': code,
'message': message
}
if details:
error_data['details'] = details
return api_response(
data={'error': error_data},
status='error',
http_code=http_code
)
def api_error(
message: str,
code: str = ErrorCodes.BAD_REQUEST,
details: Dict = None,
http_code: int = 400
):
"""
Simplified error response helper.
Args:
message: Human-readable error message
code: Error code (default BAD_REQUEST)
details: Optional error details
http_code: HTTP status code (default 400)
"""
return error_response(code=code, message=message, details=details, http_code=http_code)
def paginated_response(
items: List,
page: int,
per_page: int,
total: int,
schema=None
):
"""
Paginated list response.
Response format:
{
"status": "success",
"data": [...],
"meta": {
"pagination": {
"page": 1,
"per_page": 20,
"total": 150,
"total_pages": 8,
"has_next": true,
"has_prev": false
}
}
}
"""
total_pages = (total + per_page - 1) // per_page if per_page > 0 else 0
if schema:
items = schema.dump(items, many=True)
return api_response(
data=items,
meta={
'pagination': {
'page': page,
'per_page': per_page,
'total': total,
'total_pages': total_pages,
'has_next': page < total_pages,
'has_prev': page > 1
}
}
)