"""Conversion functions for ADK memory records.
Provides utilities for extracting searchable text from ADK Content objects
and converting between ADK models and database records.
"""
from datetime import datetime, timezone
from typing import TYPE_CHECKING, Any
from google.adk.memory.memory_entry import MemoryEntry
from google.genai import types
from sqlspec.extensions.adk.memory._types import StoredMemory
from sqlspec.utils.logging import get_logger
from sqlspec.utils.uuids import uuid4
if TYPE_CHECKING:
from collections.abc import Mapping, Sequence
from google.adk.events.event import Event
from google.adk.sessions import Session
__all__ = (
"event_to_memory_record",
"extract_content_text",
"memory_entry_to_record",
"record_to_memory_entry",
"records_to_memory_entries",
"session_to_memory_records",
)
logger = get_logger("sqlspec.extensions.adk.memory.converters")
_UNKNOWN_SESSION_ID = "__unknown_session_id__"
[docs]
def event_to_memory_record(
event: "Event",
session_id: str,
app_name: str,
user_id: str,
scope: str = "user",
embedding: "Sequence[float] | None" = None,
) -> "StoredMemory | None":
"""Convert an ADK Event to a stored memory record.
Args:
event: ADK Event object.
session_id: ID of the parent session.
app_name: Name of the application.
user_id: ID of the user.
scope: Visibility scope ('user' or 'app').
embedding: Optional vector embedding stored alongside the entry.
Returns:
StoredMemory for database storage, or None if event has no content.
"""
if event.content is None:
return None
content_text = extract_content_text(event.content)
if not content_text.strip():
return None
content_dict = event.content.model_dump(exclude_none=True, mode="json")
custom_metadata = dict(event.custom_metadata) if event.custom_metadata else None
now = datetime.now(timezone.utc)
return StoredMemory(
id=str(uuid4()),
session_id=session_id,
app_name=app_name,
user_id=user_id,
scope=scope,
event_id=event.id,
author=event.author,
timestamp=datetime.fromtimestamp(event.timestamp, tz=timezone.utc),
content_json=content_dict,
content_text=content_text,
metadata_json=custom_metadata,
inserted_at=now,
embedding=embedding,
)
[docs]
def memory_entry_to_record(
entry: "MemoryEntry",
app_name: str,
user_id: str,
extra_metadata: "dict[str, Any] | None" = None,
scope: str = "user",
embedding: "Sequence[float] | None" = None,
) -> "StoredMemory | None":
"""Convert an ADK MemoryEntry to a database record.
Serializes the entry's ``content`` to ``content_json``, extracts text
from ``content.parts`` for ``content_text``, and merges entry-level
``custom_metadata`` with the optional ``extra_metadata`` parameter.
Args:
entry: ADK MemoryEntry object.
app_name: Name of the application.
user_id: ID of the user.
extra_metadata: Optional call-level metadata to merge with the
entry's own ``custom_metadata``.
scope: Visibility scope ('user' or 'app').
embedding: Optional vector embedding stored alongside the entry.
Returns:
StoredMemory for database storage, or None if entry has no
indexable content.
"""
content_text = extract_content_text(entry.content)
if not content_text.strip():
return None
content_dict = entry.content.model_dump(exclude_none=True, mode="json")
merged_metadata: dict[str, Any] | None = None
if entry.custom_metadata or extra_metadata:
merged_metadata = {}
if extra_metadata:
merged_metadata.update(extra_metadata)
if entry.custom_metadata:
merged_metadata.update(entry.custom_metadata)
now = datetime.now(timezone.utc)
timestamp = now
if entry.timestamp:
try:
timestamp = datetime.fromisoformat(entry.timestamp)
except (ValueError, TypeError):
timestamp = now
record_id = entry.id or str(uuid4())
return StoredMemory(
id=record_id,
session_id=_UNKNOWN_SESSION_ID,
app_name=app_name,
user_id=user_id,
scope=scope,
event_id=record_id,
author=entry.author or "",
timestamp=timestamp,
content_json=content_dict,
content_text=content_text,
metadata_json=merged_metadata,
inserted_at=now,
embedding=embedding,
)
[docs]
def session_to_memory_records(
session: "Session", scope: str = "user", embeddings: "Mapping[str, Sequence[float]] | None" = None
) -> list["StoredMemory"]:
"""Convert a completed ADK Session to a list of stored memory records.
Extracts all events with content from the session and converts
them to memory records for storage.
Args:
session: ADK Session object with events.
scope: Visibility scope ('user' or 'app').
embeddings: Optional map of event id to vector embedding.
Returns:
List of StoredMemory objects for database storage.
"""
records: list[StoredMemory] = []
if not session.events:
return records
for event in session.events:
record = event_to_memory_record(
event=event,
session_id=session.id,
app_name=session.app_name,
user_id=session.user_id,
scope=scope,
embedding=embeddings.get(event.id) if embeddings else None,
)
if record is not None:
records.append(record)
return records
[docs]
def record_to_memory_entry(record: "StoredMemory") -> "MemoryEntry":
"""Convert a database record to an ADK MemoryEntry.
Preserves ``id``, ``custom_metadata``, and propagates ``scope``
into the entry's custom metadata.
Args:
record: Memory database record.
Returns:
ADK MemoryEntry object with all available fields populated.
"""
content = types.Content.model_validate(record["content_json"])
timestamp_str = record["timestamp"].isoformat() if record["timestamp"] else None
custom_metadata = dict(record["metadata_json"] or {})
if "scope" not in custom_metadata and record.get("scope"):
custom_metadata["scope"] = record["scope"]
return MemoryEntry(
id=record["id"],
content=content,
author=record["author"],
timestamp=timestamp_str,
custom_metadata=custom_metadata,
)
[docs]
def records_to_memory_entries(records: list["StoredMemory"]) -> list["Any"]:
"""Convert a list of database records to ADK MemoryEntry objects.
Args:
records: List of memory database records.
Returns:
List of ADK MemoryEntry objects.
"""
return [record_to_memory_entry(record) for record in records]
