40 lines
930 B
Python
40 lines
930 B
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from datetime import datetime
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
|
||
|
|
def _isoformat(value: Any) -> Any:
|
||
|
|
if isinstance(value, datetime):
|
||
|
|
return value.isoformat()
|
||
|
|
return value
|
||
|
|
|
||
|
|
|
||
|
|
def build_integration_meta(
|
||
|
|
*,
|
||
|
|
flow_type: str,
|
||
|
|
source_type: str,
|
||
|
|
source_service: str,
|
||
|
|
ownership: str,
|
||
|
|
live: bool,
|
||
|
|
cached: bool,
|
||
|
|
generated_at: Any = None,
|
||
|
|
snapshot_at: Any = None,
|
||
|
|
notes: list[str] | None = None,
|
||
|
|
) -> dict[str, Any]:
|
||
|
|
meta = {
|
||
|
|
"flow_type": flow_type,
|
||
|
|
"source_type": source_type,
|
||
|
|
"source_service": source_service,
|
||
|
|
"ownership": ownership,
|
||
|
|
"live": live,
|
||
|
|
"cached": cached,
|
||
|
|
}
|
||
|
|
if generated_at is not None:
|
||
|
|
meta["generated_at"] = _isoformat(generated_at)
|
||
|
|
if snapshot_at is not None:
|
||
|
|
meta["snapshot_at"] = _isoformat(snapshot_at)
|
||
|
|
if notes:
|
||
|
|
meta["notes"] = notes
|
||
|
|
return meta
|