Manual Cadwyn migrator

migrator.py

"""
Socket-compatible migration runner.

How it bypasses FastAPI/HTTP coupling
--------------------------------------
Cadwyn's VersionChange decorators store transformer callables on instruction
objects.  When an instruction is called it simply does:

    def __call__(self, info):
        return self.transformer(info)

The transformer itself only accesses `info.body` (a plain dict).  Nothing
about FastAPI's Request/Response is exercised.  We therefore pass a minimal
duck-type container — _BodyInfo — instead of the real RequestInfo/ResponseInfo.

Why there is no body_type parameter
-------------------------------------
In Cadwyn's HTTP mode the instruction dicts are keyed by the Pydantic type of
each route's request/response body, so the dispatcher can filter by type when
a single handler might serve many different schemas.

In a socket protocol the direction of each call already determines which
schema is in play: every message going *in* is a Q and every message coming
*back* is an S.  There is no ambiguity, so we simply apply all instructions
that fall within the version range, regardless of schema key.

Public API
----------
upgrade_request(body, client_version, server_version, bundle, version_values)
    Walk from client_version up to server_version, applying every request
    transformer declared in that range.

downgrade_response(body, server_version, client_version, bundle, version_values)
    Walk from server_version down to client_version, applying every response
    transformer declared in that range.
"""

from __future__ import annotations

from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    from cadwyn.structure.versions import VersionBundle


# ---------------------------------------------------------------------------
# Minimal container — replaces RequestInfo / ResponseInfo at runtime
# ---------------------------------------------------------------------------

class _BodyInfo:
    """
    Duck-type stand-in for Cadwyn's RequestInfo / ResponseInfo.

    Transformer functions decorated with convert_request_to_next_version_for
    or convert_response_to_previous_version_for receive this object.  They
    only mutate `.body`, so no HTTP plumbing is needed.
    """
    __slots__ = ("body",)

    def __init__(self, body: dict[str, Any]) -> None:
        self.body = dict(body)   # copy so the original is not mutated in place


# ---------------------------------------------------------------------------
# Migration runner
# ---------------------------------------------------------------------------

def upgrade_request(
    body: dict[str, Any],
    client_version: str,
    server_version: str,
    bundle: "VersionBundle",
    version_values: dict[str, str],
) -> dict[str, Any]:
    """
    Upgrade *body* from *client_version* up to *server_version*.

    Iterates bundle.reversed_versions (oldest → newest) and applies every
    request transformer that falls strictly between the two version boundaries.
    """
    client_val = version_values[client_version]
    server_val = version_values[server_version]

    info = _BodyInfo(body)
    for version in bundle.reversed_versions:          # oldest → newest
        if version.value <= client_val:
            continue                                  # already at or before client
        if version.value > server_val:
            continue                                  # beyond this server's head
        for change in version.changes:
            for instructions in change.alter_request_by_schema_instructions.values():
                for instr in instructions:
                    instr(info)

    return info.body


def downgrade_response(
    body: dict[str, Any],
    server_version: str,
    client_version: str,
    bundle: "VersionBundle",
    version_values: dict[str, str],
) -> dict[str, Any]:
    """
    Downgrade *body* from *server_version* down to *client_version*.

    Iterates bundle.versions (newest → oldest) and applies every response
    transformer that falls strictly above the client boundary.
    """
    client_val = version_values[client_version]
    server_val = version_values[server_version]

    info = _BodyInfo(body)
    for version in bundle.versions:                   # newest → oldest
        if version.value > server_val:
            continue                                  # above this server's head
        if version.value <= client_val:
            break                                     # reached the client's version
        for change in version.changes:
            for instructions in change.alter_response_by_schema_instructions.values():
                for instr in instructions:
                    instr(info)

    return info.body