tegwick
388b330809
Share LegacyWorkstreamIdBodyMixin across create schemas; meter POST /tasks/ and /decisions/ workstream_id bodies. State summary uses workplan flow; NextStep dual-writes workplan_* fields alongside legacy workstream_*.
111 lines
No EOL
3.4 KiB
Python
111 lines
No EOL
3.4 KiB
Python
"""Shared legacy REST compat metering (routes and query params)."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from fastapi import Request, Response
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from api.services.legacy_meter import identity_from_request, record_legacy_usage
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_LEGACY_OWNER = "state-hub.api"
|
|
# Planning horizon for legacy workstream interfaces; actual removal remains
|
|
# legacy-meter-gated per docs/workplan-terminology-transition.md.
|
|
_LEGACY_WORKSTREAM_SUNSET = "Wed, 30 Jun 2027 23:59:59 GMT"
|
|
|
|
|
|
def mark_legacy_response(response: Response | None, replacement_ref: str) -> None:
|
|
if response is None:
|
|
return
|
|
response.headers["Deprecation"] = "true"
|
|
response.headers["Sunset"] = _LEGACY_WORKSTREAM_SUNSET
|
|
response.headers["X-StateHub-Replacement"] = replacement_ref
|
|
response.headers.append("Link", f'<{replacement_ref}>; rel="successor-version"')
|
|
|
|
|
|
def legacy_query_param_key(method: str, route: str, param: str = "workstream_id") -> str:
|
|
return f"rest_api:{method} {route}?{param}"
|
|
|
|
|
|
async def meter_legacy_query_param(
|
|
*,
|
|
session: AsyncSession,
|
|
request: Request | None,
|
|
response: Response | None,
|
|
method: str,
|
|
route: str,
|
|
replacement_ref: str,
|
|
param: str = "workstream_id",
|
|
) -> None:
|
|
interface_key = legacy_query_param_key(method, route, param)
|
|
mark_legacy_response(response, replacement_ref)
|
|
try:
|
|
await record_legacy_usage(
|
|
session,
|
|
interface_key=interface_key,
|
|
interface_kind="rest_api",
|
|
replacement_ref=replacement_ref,
|
|
owner_component=_LEGACY_OWNER,
|
|
replacement_verified=True,
|
|
identity=identity_from_request(request),
|
|
)
|
|
except Exception:
|
|
await session.rollback()
|
|
logger.warning("legacy-meter failed to record %s", interface_key, exc_info=True)
|
|
|
|
|
|
def legacy_body_field_key(method: str, route: str, field: str = "workstream_id") -> str:
|
|
return f"rest_api:{method} {route} {field}"
|
|
|
|
|
|
async def meter_legacy_body_field(
|
|
*,
|
|
session: AsyncSession,
|
|
request: Request | None,
|
|
response: Response | None,
|
|
method: str,
|
|
route: str,
|
|
replacement_ref: str,
|
|
field: str = "workstream_id",
|
|
) -> None:
|
|
interface_key = legacy_body_field_key(method, route, field)
|
|
mark_legacy_response(response, replacement_ref)
|
|
try:
|
|
await record_legacy_usage(
|
|
session,
|
|
interface_key=interface_key,
|
|
interface_kind="rest_api",
|
|
replacement_ref=replacement_ref,
|
|
owner_component=_LEGACY_OWNER,
|
|
replacement_verified=True,
|
|
identity=identity_from_request(request),
|
|
)
|
|
except Exception:
|
|
await session.rollback()
|
|
logger.warning("legacy-meter failed to record %s", interface_key, exc_info=True)
|
|
|
|
|
|
async def meter_legacy_body_from_model(
|
|
body: object,
|
|
*,
|
|
session: AsyncSession,
|
|
request: Request | None,
|
|
response: Response | None,
|
|
method: str,
|
|
route: str,
|
|
replacement_ref: str,
|
|
field: str = "workstream_id",
|
|
) -> None:
|
|
if not getattr(body, "used_legacy_workstream_id", False):
|
|
return
|
|
await meter_legacy_body_field(
|
|
session=session,
|
|
request=request,
|
|
response=response,
|
|
method=method,
|
|
route=route,
|
|
replacement_ref=replacement_ref,
|
|
field=field,
|
|
) |