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

from shopdb.core.models import AuditLog


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__ = ['audit_log', 'resolve_asset_position']