shopdb-flask/shopdb/api/__init__.py

"""Public API namespace exposed to plugins.

Plugin authors at sister sites import from this module. The contract is
locked in ADR-001 and versioned per ADR-002. Helpers added here become
part of the platform contract; bumps follow ADR-002 rules.

Currently exposed:
- audit_log: record an audit log entry with consistent schema
- resolve_asset_position: compute an asset's resolved map position

Setting helpers are exposed via BasePlugin instance methods
(plugin.get_setting, plugin.set_setting), not from this namespace.
"""

from typing import Any, Dict, Optional

# -- Plugin contract surface (ADR-001, versioned per ADR-002) ----------------
# Everything a plugin is allowed to import from the core lives here. Plugins
# import these from `shopdb.api`, never from internal paths like
# `shopdb.core.models.*` or `shopdb.extensions`. The contract test
# (tests/test_plugin_contract.py) enforces this. Adding a name here is an
# additive (minor) contract change; removing one is breaking (major).

# Infrastructure
from shopdb.extensions import db, cache

# Model base classes for declaring plugin tables
from shopdb.core.models.base import BaseModel, AuditMixin

# Core domain models plugins legitimately reference (the asset contract)
from shopdb.core.models import (
    Asset,
    AssetType,
    AssetStatus,
    Vendor,
    Model,
    Communication,
    CommunicationType,
    Location,
    Setting,
    AuditLog,
    Application,
    AppVersion,
    OperatingSystem,
    AssetRelationship,
    RelationshipType,
    User,
    Role,
    SupportTeam,
)
# Display-role mapping (which kiosk shows which surface). On the surface
# because a plugin reporting on displays has no other way to name what a
# display IS - the role lives in core, not in any plugin.
from shopdb.core.models.dashboarddefault import (
    DashboardDefault,
    DISPLAY_ROLES,
    DISPLAY_ROLE_PATHS,
    normalize_display_role,
)

# Response + pagination helpers for plugin API blueprints
from shopdb.utils.responses import (
    success_response,
    error_response,
    paginated_response,
    ErrorCodes,
)
from shopdb.utils.pagination import get_pagination_params, paginate_query

# Authorization decorators for gating plugin write routes
from shopdb.utils.authz import require_permission, require_role

# Service-token authorization for unattended plugin endpoints (collector,
# GE-Enforce fetch, ...): checks a scoped managed token without exposing token
# internals.
from shopdb.utils.apitoken_auth import (
    service_token_authorized, authorized_service_token,
)

# Import-mode helpers: preserve legacy timestamps during a bulk data import
from shopdb.utils.import_mode import (
    apply_import_timestamps,
    import_mode_active,
    parse_import_datetime,
)

# Dualpath single-machine collapse (a dual-bay pair is one physical machine)
from shopdb.core.services.dualpath import (
    resolve_dualpath_pairs,
    dualpath_single_machine_enabled,
)

# Legacy employee directory lookup (read-only) used by notifications
from shopdb.utils.employee_db import employee_connection
from shopdb.utils.mailer import send_email, send_alert, send_webhook

# CMMC USB check-in/out database (read-write) used by the usb plugin
from shopdb.utils.cmmc_usb_db import cmmc_usb_connection


def audit_log(
    action: str,
    entitytype: str,
    entityid: int = None,
    entityname: str = None,
    changes: Dict = None,
    details: Dict = None,
) -> AuditLog:
    """Record an audit log entry with the framework's standard schema.

    Plugins call this to record state changes on their own assets in a
    way that is consistent with core auditing. The function delegates to
    AuditLog.log() which already captures the current user, IP address,
    and user agent from the Flask request context.

    Args:
        action: Action verb in past tense ('created', 'updated', 'deleted')
        entitytype: Class name of the entity affected ('Computer', 'Printer')
        entityid: Primary key of the entity
        entityname: Human-readable identifier (hostname, asset number)
        changes: Dict with 'before' and 'after' snapshots for updates
        details: Arbitrary additional context

    Returns:
        The created AuditLog instance, already committed to the DB.
    """
    return AuditLog.log(
        action=action,
        entitytype=entitytype,
        entityid=entityid,
        entityname=entityname,
        changes=changes,
        details=details,
    )


# Cap how deep the relationship-walk traverses before giving up. ADR-001
# specifies a max walk depth of 3 to bound the work per request, with a
# visited-set guarding against cycles. Past this depth the walk treats the
# next hop as if it had no position to contribute.
_POSITION_WALK_MAX_DEPTH = 3

# Relationship type names whose edges are eligible for the inheritance walk
# when inheritsposition is true on the edge. Ordered by priority per
# ADR-001 ("partof first, then controls"). Edges of other types are never
# followed even if inheritsposition is true.
_INHERITABLE_TYPES = ('partof', 'controls')


def _walk_related_for_position(asset, visited, depth):
    """Recursive helper for resolve_asset_position relationship walk. Returns
    a (mapx, mapy) tuple from the first related asset whose position
    resolves, or None. Visited tracks assetids already explored to break
    cycles."""
    if depth >= _POSITION_WALK_MAX_DEPTH:
        return None
    aid = getattr(asset, 'assetid', None)
    if aid is None or aid in visited:
        return None
    visited.add(aid)

    edges = []
    for rel in list(getattr(asset, 'outgoing_relationships', []) or []):
        edges.append((rel, getattr(rel, 'targetasset', None)))
    for rel in list(getattr(asset, 'incoming_relationships', []) or []):
        edges.append((rel, getattr(rel, 'sourceasset', None)))

    def _priority(edge):
        rel = edge[0]
        rtype = getattr(rel, 'relationshiptype', None)
        type_name = getattr(rtype, 'relationshiptype', '') if rtype else ''
        try:
            return _INHERITABLE_TYPES.index(type_name)
        except ValueError:
            return len(_INHERITABLE_TYPES)

    edges.sort(key=_priority)

    for rel, neighbor in edges:
        if neighbor is None:
            continue
        if not getattr(rel, 'inheritsposition', False):
            continue
        if not getattr(rel, 'isactive', True):
            continue
        rtype = getattr(rel, 'relationshiptype', None)
        type_name = getattr(rtype, 'relationshiptype', '') if rtype else ''
        if type_name not in _INHERITABLE_TYPES:
            continue

        n_mapx = getattr(neighbor, 'mapx', None)
        n_mapy = getattr(neighbor, 'mapy', None)
        if n_mapx is not None and n_mapy is not None:
            return (n_mapx, n_mapy)

        recursed = _walk_related_for_position(neighbor, visited, depth + 1)
        if recursed is not None:
            return recursed
    return None


def resolve_asset_position(asset) -> Optional[Dict[str, Any]]:
    """Compute the resolved map position for an asset.

    Per ADR-001, position resolution follows this priority chain:
      1. Asset-specific override (asset.mapx, asset.mapy)
      2. Walk relationships where inheritsposition is true on edges of type
         partof or controls (partof first), depth-limited and cycle-safe
      3. Asset's location coords (asset.location.mapx, .mapy)
      4. None (asset is unplaced, rendered in a tray)

    Returns a dict {'mapx', 'mapy', 'positionsource'} where positionsource
    is one of 'self', 'related', 'location'. Returns None when no priority
    yields coordinates.
    """
    mapx = getattr(asset, 'mapx', None)
    mapy = getattr(asset, 'mapy', None)
    if mapx is not None and mapy is not None:
        return {'mapx': mapx, 'mapy': mapy, 'positionsource': 'self'}

    related = _walk_related_for_position(asset, set(), 0)
    if related is not None:
        return {'mapx': related[0], 'mapy': related[1], 'positionsource': 'related'}

    location = getattr(asset, 'location', None)
    if location is not None:
        loc_mapx = getattr(location, 'mapx', None)
        loc_mapy = getattr(location, 'mapy', None)
        if loc_mapx is not None and loc_mapy is not None:
            return {
                'mapx': loc_mapx,
                'mapy': loc_mapy,
                'positionsource': 'location',
            }

    return None


__all__ = [
    # Helpers
    'audit_log',
    'resolve_asset_position',
    'resolve_dualpath_pairs',
    'dualpath_single_machine_enabled',
    # Infrastructure
    'db',
    'cache',
    # Model bases
    'BaseModel',
    'AuditMixin',
    # Core models
    'Asset',
    'AssetType',
    'AssetStatus',
    'Vendor',
    'Model',
    'Communication',
    'CommunicationType',
    'Location',
    'Setting',
    'AuditLog',
    'Application',
    'AppVersion',
    'OperatingSystem',
    'AssetRelationship',
    'RelationshipType',
    # Display roles. The role vocabulary is the kiosk's own, so a plugin that
    # reads a reported display type resolves it the same way core does.
    'DashboardDefault',
    'DISPLAY_ROLES',
    'DISPLAY_ROLE_PATHS',
    'normalize_display_role',
    # Response + pagination helpers
    'success_response',
    'error_response',
    'paginated_response',
    'ErrorCodes',
    'get_pagination_params',
    'paginate_query',
    # Authorization decorators
    'require_permission',
    'require_role',
    'service_token_authorized',
    'authorized_service_token',
    'SupportTeam',
    # Import-mode helpers
    'apply_import_timestamps',
    'import_mode_active',
    'parse_import_datetime',
    # Legacy employee directory
    'employee_connection',
    'send_email',
    'send_alert',
    'send_webhook',
    'User',
    'Role',
    # CMMC USB check-in/out database
    'cmmc_usb_connection',
]