"""Base store class for ADK session backends."""
import inspect
import logging
from abc import ABC, abstractmethod
from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING, Any, Final, Generic, TypeVar, cast
from sqlspec.extensions.adk._config_utils import _adk_session_store_config
from sqlspec.extensions.adk._table_utils import ensure_table_name, owner_id_column_name, unique_statements
from sqlspec.migrations.schema import SchemaTarget, ensure_schema_async, ensure_schema_sync
from sqlspec.observability import resolve_db_system
from sqlspec.utils.logging import get_logger, log_with_context
from sqlspec.utils.sync_tools import async_
if TYPE_CHECKING:
from collections.abc import Callable
from sqlspec.config import DatabaseConfigProtocol
from sqlspec.extensions.adk._types import StoredEvent, StoredSession
__all__ = ("BaseAsyncADKStore", "BaseSyncADKStore")
ConfigT = TypeVar("ConfigT", bound="DatabaseConfigProtocol[Any, Any, Any]")
logger = get_logger("sqlspec.extensions.adk.store")
ADK_RESET_TABLE_PROFILES: Final = (
("adk_session", "adk_event", "adk_app_state", "adk_user_state", "adk_internal_metadata"),
("adk_session", "adk_event", "adk_app_state", "adk_user_state", "adk_metadata"),
("adk_sessions", "adk_events", "adk_app_states", "adk_user_states", "adk_internal_metadata"),
("adk_sessions", "adk_events", "adk_app_states", "adk_user_states", "adk_metadata"),
)
class _ADKStoreCommon(Generic[ConfigT]):
"""Shared non-async ADK store state and helpers."""
if TYPE_CHECKING:
_drop_tables_sql: "Callable[[], list[str]]"
__slots__ = (
"_app_state_table",
"_config",
"_events_table",
"_metadata_table",
"_owner_id_column_ddl",
"_owner_id_column_name",
"_session_table",
"_user_state_table",
)
def __init__(self, config: ConfigT) -> None:
"""Initialize the ADK store.
Args:
config: SQLSpec database configuration.
Notes:
Reads configuration from config.extension_config["adk"]:
- session_table: Sessions table name (default: "adk_session")
- events_table: Events table name (default: "adk_event")
- app_state_table: App-scoped state table name (default: "adk_app_state")
- user_state_table: User-scoped state table name (default: "adk_user_state")
- metadata_table: Internal metadata table name (default: "adk_internal_metadata")
- owner_id_column: Optional owner FK column DDL (default: None)
"""
self._config = config
store_config = self._store_config_from_extension()
self._session_table: str = str(store_config["session_table"])
self._events_table: str = str(store_config["events_table"])
self._app_state_table: str = str(store_config["app_state_table"])
self._user_state_table: str = str(store_config["user_state_table"])
self._metadata_table: str = str(store_config["metadata_table"])
self._owner_id_column_ddl: str | None = store_config.get("owner_id_column")
self._owner_id_column_name: str | None = (
owner_id_column_name(self._owner_id_column_ddl) if self._owner_id_column_ddl else None
)
ensure_table_name(self._session_table)
ensure_table_name(self._events_table)
ensure_table_name(self._app_state_table)
ensure_table_name(self._user_state_table)
ensure_table_name(self._metadata_table)
@property
def config(self) -> ConfigT:
"""Return the database configuration."""
return self._config
@property
def session_table(self) -> str:
"""Return the sessions table name."""
return self._session_table
@property
def events_table(self) -> str:
"""Return the events table name."""
return self._events_table
@property
def app_state_table(self) -> str:
"""Return the app-scoped state table name."""
return self._app_state_table
@property
def user_state_table(self) -> str:
"""Return the user-scoped state table name."""
return self._user_state_table
@property
def metadata_table(self) -> str:
"""Return the ADK metadata table name."""
return self._metadata_table
@property
def owner_id_column_ddl(self) -> "str | None":
"""Return the full owner ID column DDL (or None if not configured)."""
return self._owner_id_column_ddl
@property
def owner_id_column_name(self) -> "str | None":
"""Return the owner ID column name only (or None if not configured)."""
return self._owner_id_column_name
@property
def create_schema_enabled(self) -> bool:
"""Return whether adapter-level table creation should run."""
manage_schema, create_schema = self._schema_management_flags()
return manage_schema and create_schema
def _reset_drop_tables_sql(self) -> "list[str]":
"""Return all table drops needed before recreating the clean-break schema."""
statements = list(self._drop_tables_sql())
for table_profile in ADK_RESET_TABLE_PROFILES:
statements.extend(self._drop_sql_for_table_profile(table_profile))
return unique_statements(statements)
def _store_config_from_extension(self) -> "dict[str, Any]":
"""Extract ADK store configuration from config.extension_config.
Returns:
Dict with ADK table names and optionally owner_id_column.
"""
return dict(_adk_session_store_config(self._config))
def _schema_management_flags(self) -> "tuple[bool, bool]":
"""Return automatic-management and missing-table creation flags."""
extension_config = cast("dict[str, Any]", self._config.extension_config)
settings = cast("dict[str, Any]", extension_config.get("adk", {}))
return bool(settings.get("manage_schema", True)), bool(settings.get("create_schema", True))
def _calculate_expires_at(self, expires_in: "int | timedelta | None") -> "datetime | None":
"""Calculate expiration timestamp from expires_in.
Args:
expires_in: Seconds or timedelta until expiration.
Returns:
UTC datetime of expiration, or None if no expiration.
"""
if expires_in is None:
return None
expires_in_seconds = int(expires_in.total_seconds()) if isinstance(expires_in, timedelta) else expires_in
if expires_in_seconds <= 0:
return None
return datetime.now(timezone.utc) + timedelta(seconds=expires_in_seconds)
def _drop_sql_for_table_profile(self, table_profile: "tuple[str, str, str, str, str]") -> "list[str]":
session_table, events_table, app_state_table, user_state_table, metadata_table = table_profile
current_session_table = self._session_table
current_events_table = self._events_table
current_app_state_table = self._app_state_table
current_user_state_table = self._user_state_table
current_table = self._metadata_table
self._session_table = session_table
self._events_table = events_table
self._app_state_table = app_state_table
self._user_state_table = user_state_table
self._metadata_table = metadata_table
try:
return list(self._drop_tables_sql())
finally:
self._session_table = current_session_table
self._events_table = current_events_table
self._app_state_table = current_app_state_table
self._user_state_table = current_user_state_table
self._metadata_table = current_table
def _log_tables_created(self) -> None:
log_with_context(
logger,
logging.DEBUG,
"adk.tables.ready",
db_system=resolve_db_system(type(self).__name__),
session_table=self._session_table,
events_table=self._events_table,
)
def _log_tables_dropped(self) -> None:
log_with_context(
logger,
logging.DEBUG,
"adk.tables.dropped",
db_system=resolve_db_system(type(self).__name__),
session_table=self._session_table,
events_table=self._events_table,
)
def _log_tables_recreated(self) -> None:
log_with_context(
logger,
logging.DEBUG,
"adk.tables.recreated",
db_system=resolve_db_system(type(self).__name__),
session_table=self._session_table,
events_table=self._events_table,
)
[docs]
class BaseAsyncADKStore(_ADKStoreCommon[ConfigT], ABC):
"""Base class for async SQLSpec-backed ADK session stores.
Implements storage operations for Google ADK sessions and events using
SQLSpec database adapters with async/await.
This abstract base class provides common functionality for all database-specific
store implementations including:
- Connection management via SQLSpec configs
- Table name validation
- Session and event CRUD operations
Subclasses must implement dialect-specific SQL queries and will be created
in each adapter directory (e.g., sqlspec/adapters/asyncpg/adk/store.py).
Args:
config: SQLSpec database configuration with extension_config["adk"] settings.
Notes:
Configuration is read from config.extension_config["adk"]:
- session_table: Sessions table name (default: "adk_session")
- events_table: Events table name (default: "adk_event")
- app_state_table: App-scoped state table name (default: "adk_app_state")
- user_state_table: User-scoped state table name (default: "adk_user_state")
- metadata_table: Internal metadata table name (default: "adk_internal_metadata")
- owner_id_column: Optional owner FK column DDL (default: None)
"""
__slots__ = ()
[docs]
async def create_tables(self) -> None:
"""Create the sessions and events tables if they don't exist."""
raise NotImplementedError
[docs]
async def prepare_schema_async(self, driver: Any) -> None:
"""Prepare adapter-specific schema decisions with an asynchronous driver."""
[docs]
async def create_session(
self, session_id: str, app_name: str, user_id: str, state: "dict[str, Any]", owner_id: "Any | None" = None
) -> "StoredSession":
"""Create a new session.
Args:
session_id: Unique identifier for the session.
app_name: Name of the application.
user_id: ID of the user.
state: Session state dictionary.
owner_id: Optional owner ID value for owner_id_column (if configured).
Returns:
The created session record.
"""
raise NotImplementedError
[docs]
@abstractmethod
async def get_session(
self, app_name: str, user_id: str, session_id: str, *, renew_for: "int | timedelta | None" = None
) -> "StoredSession | None":
"""Get a session.
Args:
app_name: Name of the application.
user_id: ID of the user.
session_id: Session identifier.
renew_for: If positive, touch the session update timestamp while reading.
Returns:
Session record if found, None otherwise.
"""
raise NotImplementedError
[docs]
@abstractmethod
async def update_session_state(self, app_name: str, user_id: str, session_id: str, state: "dict[str, Any]") -> None:
"""Update session state.
Args:
app_name: Name of the application.
user_id: ID of the user.
session_id: Session identifier.
state: New state dictionary.
"""
raise NotImplementedError
[docs]
@abstractmethod
async def list_sessions(self, app_name: str, user_id: "str | None" = None) -> "list[StoredSession]":
"""List all sessions for an app, optionally filtered by user.
Args:
app_name: Name of the application.
user_id: ID of the user. If None, returns all sessions for the app.
Returns:
List of session records.
"""
raise NotImplementedError
[docs]
@abstractmethod
async def delete_session(self, app_name: str, user_id: str, session_id: str) -> None:
"""Delete a session and its events.
Args:
app_name: Name of the application.
user_id: ID of the user.
session_id: Session identifier.
"""
raise NotImplementedError
[docs]
@abstractmethod
async def append_event(self, event_record: "StoredEvent") -> None:
"""Append an event to a session.
Args:
event_record: Event record to insert.
"""
raise NotImplementedError
[docs]
@abstractmethod
async def append_event_and_update_state(
self,
event_record: "StoredEvent",
app_name: str,
user_id: str,
session_id: str,
state: "dict[str, Any]",
*,
app_state: "dict[str, Any] | None" = None,
user_state: "dict[str, Any] | None" = None,
) -> "StoredSession":
"""Atomically append an event and update the session's durable state.
This is the authoritative durable write boundary for post-creation
app-scoped snapshot to replace/upsert for ``app_name``. When
``user_state`` is provided, it is a full merged user-scoped snapshot to
replace/upsert for ``(app_name, user_id)``. ``None`` means that scope
was untouched by the event and must not be written.
Args:
event_record: Event record to store.
app_name: Application name for routing scoped-state upserts.
user_id: User identifier for routing user-scoped upserts.
session_id: Session identifier whose state should be updated.
state: Post-append durable session-scoped state snapshot
(``temp:`` keys already stripped by the service layer).
app_state: Full app-scoped state snapshot (``app:*`` keys) to
upsert atomically, or ``None`` when untouched.
user_state: Full user-scoped state snapshot (``user:*`` keys) to
upsert atomically, or ``None`` when untouched.
Returns:
The updated StoredSession reflecting the new state and update_time.
Raises:
ValueError: If the session row no longer exists at update time
(raced with delete_session).
"""
raise NotImplementedError
[docs]
@abstractmethod
async def get_events(
self,
app_name: str,
user_id: str,
session_id: str,
after_timestamp: "datetime | None" = None,
limit: "int | None" = None,
) -> "list[StoredEvent]":
"""Get events for a session.
Args:
app_name: Name of the application.
user_id: ID of the user.
session_id: Session identifier.
after_timestamp: Only return events after this time.
limit: Maximum number of events to return.
Returns:
List of event records ordered by timestamp ascending.
"""
raise NotImplementedError
[docs]
@abstractmethod
async def delete_expired_events(self, before: datetime, app_name: "str | None" = None) -> int:
"""Delete events older than the given timestamp.
Args:
before: Timestamp threshold; events with timestamp earlier than this value are deleted.
app_name: When given, restrict deletion to events belonging to this application.
Returns:
Number of event rows deleted.
"""
raise NotImplementedError
[docs]
@abstractmethod
async def delete_idle_user_states(self, updated_before: datetime, app_name: "str | None" = None) -> int:
"""Delete user-scoped state rows whose update_time predates the given threshold.
Args:
updated_before: Timestamp threshold; rows updated earlier than this value are deleted.
app_name: When given, restrict deletion to rows belonging to this application.
Returns:
Number of user state rows deleted.
"""
raise NotImplementedError
[docs]
@abstractmethod
async def delete_idle_sessions(self, updated_before: datetime, app_name: "str | None" = None) -> int:
"""Delete sessions whose update_time predates the given threshold.
Args:
updated_before: Timestamp threshold; sessions updated earlier than this value are deleted.
app_name: When given, restrict deletion to sessions belonging to this application.
Returns:
Number of session rows deleted.
"""
raise NotImplementedError
[docs]
@abstractmethod
async def get_app_state(self, app_name: str) -> "dict[str, Any] | None":
"""Return app-scoped state for an application.
Args:
app_name: Application name.
Returns:
App-scoped state mapping if present, otherwise ``None``.
"""
raise NotImplementedError
[docs]
@abstractmethod
async def get_user_state(self, app_name: str, user_id: str) -> "dict[str, Any] | None":
"""Return user-scoped state for an application user.
Args:
app_name: Application name.
user_id: User identifier.
Returns:
User-scoped state mapping if present, otherwise ``None``.
"""
raise NotImplementedError
[docs]
@abstractmethod
async def upsert_app_state(self, app_name: str, state: "dict[str, Any]") -> None:
"""Insert or replace app-scoped state for an application.
Args:
app_name: Application name.
state: App-scoped state mapping.
"""
raise NotImplementedError
[docs]
@abstractmethod
async def upsert_user_state(self, app_name: str, user_id: str, state: "dict[str, Any]") -> None:
"""Insert or replace user-scoped state for an application user.
Args:
app_name: Application name.
user_id: User identifier.
state: User-scoped state mapping.
"""
raise NotImplementedError
[docs]
async def reconcile_schema(self, *, assume_existing: bool = False) -> None:
"""Apply additive ADK table changes from canonical adapter DDL.
Args:
assume_existing: Skip table discovery after adapter-level creation.
"""
manage_schema, create_schema = self._schema_management_flags()
if not manage_schema:
return
statement_config = getattr(self._config, "statement_config", None)
dialect = getattr(statement_config, "dialect", None)
ddl_targets = (
(self._session_table, await self._sessions_table_ddl()),
(self._events_table, await self._events_table_ddl()),
(self._app_state_table, await self._app_states_table_ddl()),
(self._user_state_table, await self._user_states_table_ddl()),
(self._metadata_table, await self._metadata_table_ddl()),
)
targets = [SchemaTarget.from_ddl(table, ddl, dialect=dialect) for table, ddl in ddl_targets]
session_context = self._config.provide_session()
if hasattr(session_context, "__aenter__"):
async with cast("Any", session_context) as driver:
await ensure_schema_async(
driver, targets, manage_schema=True, create_schema=create_schema, assume_existing=assume_existing
)
return
await async_(_reconcile_adk_schema_sync)(
self._config, targets, create_schema=create_schema, assume_existing=assume_existing
)
[docs]
async def ensure_tables(self) -> None:
"""Create tables and emit a standardized log entry."""
if self.create_schema_enabled:
await self.create_tables()
await self.reconcile_schema(assume_existing=self.create_schema_enabled)
self._log_tables_created()
[docs]
async def drop_tables(self) -> None:
"""Drop all ADK tables managed by this store in FK-safe order."""
await self._execute_lifecycle_scripts(self._drop_tables_sql())
self._log_tables_dropped()
[docs]
async def recreate_tables(self) -> None:
"""Drop and recreate all ADK tables managed by this store."""
await self.drop_tables()
await self.ensure_tables()
self._log_tables_recreated()
async def _execute_lifecycle_scripts(self, statements: list[str]) -> None:
"""Execute lifecycle DDL scripts for async and sync-backed configs."""
session_context = self._config.provide_session()
if hasattr(session_context, "__aenter__"):
async with cast("Any", session_context) as driver:
for statement in statements:
result = driver.execute_script(statement)
if inspect.isawaitable(result):
await result
commit = getattr(driver, "commit", None)
if callable(commit):
result = commit()
if inspect.isawaitable(result):
await result
return
await async_(_run_lifecycle_sync)(self._config, statements)
@abstractmethod
async def _sessions_table_ddl(self) -> str:
"""Get the CREATE TABLE SQL for the sessions table.
Returns:
SQL statement to create the sessions table.
"""
raise NotImplementedError
@abstractmethod
async def _events_table_ddl(self) -> str:
"""Get the CREATE TABLE SQL for the events table.
Returns:
SQL statement to create the events table.
"""
raise NotImplementedError
@abstractmethod
async def _app_states_table_ddl(self) -> str:
"""Get the CREATE TABLE SQL for the app-scoped state table.
Returns:
SQL statement to create the app-scoped state table.
"""
raise NotImplementedError
@abstractmethod
async def _user_states_table_ddl(self) -> str:
"""Get the CREATE TABLE SQL for the user-scoped state table.
Returns:
SQL statement to create the user-scoped state table.
"""
raise NotImplementedError
@abstractmethod
async def _metadata_table_ddl(self) -> str:
"""Get the CREATE TABLE SQL for the ADK internal metadata table.
Returns:
SQL statement to create the ADK internal metadata table.
"""
raise NotImplementedError
@abstractmethod
def _drop_app_states_table_sql(self) -> str:
"""Get the DROP TABLE SQL statement for the app-scoped state table.
Returns:
SQL statement to drop the app-scoped state table.
"""
raise NotImplementedError
@abstractmethod
def _drop_user_states_table_sql(self) -> str:
"""Get the DROP TABLE SQL statement for the user-scoped state table.
Returns:
SQL statement to drop the user-scoped state table.
"""
raise NotImplementedError
@abstractmethod
def _drop_metadata_table_sql(self) -> str:
"""Get the DROP TABLE SQL statement for the ADK internal metadata table.
Returns:
SQL statement to drop the ADK internal metadata table.
"""
raise NotImplementedError
@abstractmethod
def _drop_tables_sql(self) -> "list[str]":
"""Get the DROP TABLE SQL statements for this database dialect.
Returns:
List of SQL statements to drop the tables and all indexes.
Order matters: drop events table before sessions table due to FK.
Notes:
Should use IF EXISTS or dialect-specific error handling
to allow idempotent migrations.
"""
raise NotImplementedError
[docs]
class BaseSyncADKStore(_ADKStoreCommon[ConfigT], ABC):
"""Base class for sync SQLSpec-backed ADK session stores.
Sync-backed adapters expose a real synchronous API for direct use in
synchronous applications. Async bridging belongs in ``SQLSpecSessionService``
when Google ADK calls into a sync store from its async service surface.
Args:
config: SQLSpec database configuration with extension_config["adk"] settings.
"""
__slots__ = ()
[docs]
@abstractmethod
def create_tables(self) -> None:
"""Create the sessions and events tables if they don't exist."""
raise NotImplementedError
[docs]
def prepare_schema_sync(self, driver: Any) -> None:
"""Prepare adapter-specific schema decisions with a synchronous driver."""
[docs]
@abstractmethod
def create_session(
self, session_id: str, app_name: str, user_id: str, state: "dict[str, Any]", owner_id: "Any | None" = None
) -> "StoredSession":
"""Create a new session."""
raise NotImplementedError
[docs]
@abstractmethod
def get_session(
self, app_name: str, user_id: str, session_id: str, *, renew_for: "int | timedelta | None" = None
) -> "StoredSession | None":
"""Get a session."""
raise NotImplementedError
[docs]
@abstractmethod
def update_session_state(self, app_name: str, user_id: str, session_id: str, state: "dict[str, Any]") -> None:
"""Update session state."""
raise NotImplementedError
[docs]
@abstractmethod
def list_sessions(self, app_name: str, user_id: "str | None" = None) -> "list[StoredSession]":
"""List all sessions for an app, optionally filtered by user."""
raise NotImplementedError
[docs]
@abstractmethod
def delete_session(self, app_name: str, user_id: str, session_id: str) -> None:
"""Delete a session and its events."""
raise NotImplementedError
[docs]
@abstractmethod
def append_event(self, event_record: "StoredEvent") -> None:
"""Append an event to a session."""
raise NotImplementedError
[docs]
@abstractmethod
def append_event_and_update_state(
self,
event_record: "StoredEvent",
app_name: str,
user_id: str,
session_id: str,
state: "dict[str, Any]",
*,
app_state: "dict[str, Any] | None" = None,
user_state: "dict[str, Any] | None" = None,
) -> "StoredSession":
"""Atomically append an event and update the session's durable state."""
raise NotImplementedError
[docs]
@abstractmethod
def get_events(
self,
app_name: str,
user_id: str,
session_id: str,
after_timestamp: "datetime | None" = None,
limit: "int | None" = None,
) -> "list[StoredEvent]":
"""Get events for a session."""
raise NotImplementedError
[docs]
@abstractmethod
def delete_expired_events(self, before: datetime, app_name: "str | None" = None) -> int:
"""Delete events older than the given timestamp, optionally scoped to one application."""
raise NotImplementedError
[docs]
@abstractmethod
def delete_idle_user_states(self, updated_before: datetime, app_name: "str | None" = None) -> int:
"""Delete user-scoped state rows whose update_time predates the threshold, optionally scoped to one application."""
raise NotImplementedError
[docs]
@abstractmethod
def delete_idle_sessions(self, updated_before: datetime, app_name: "str | None" = None) -> int:
"""Delete sessions whose update_time predates the given threshold, optionally scoped to one application."""
raise NotImplementedError
[docs]
@abstractmethod
def get_app_state(self, app_name: str) -> "dict[str, Any] | None":
"""Return app-scoped state for an application."""
raise NotImplementedError
[docs]
@abstractmethod
def get_user_state(self, app_name: str, user_id: str) -> "dict[str, Any] | None":
"""Return user-scoped state for an application user."""
raise NotImplementedError
[docs]
@abstractmethod
def upsert_app_state(self, app_name: str, state: "dict[str, Any]") -> None:
"""Insert or replace app-scoped state for an application."""
raise NotImplementedError
[docs]
@abstractmethod
def upsert_user_state(self, app_name: str, user_id: str, state: "dict[str, Any]") -> None:
"""Insert or replace user-scoped state for an application user."""
raise NotImplementedError
[docs]
def reconcile_schema(self, *, assume_existing: bool = False) -> None:
"""Apply additive ADK table changes from canonical adapter DDL.
Args:
assume_existing: Skip table discovery after adapter-level creation.
"""
manage_schema, create_schema = self._schema_management_flags()
if not manage_schema:
return
statement_config = getattr(self._config, "statement_config", None)
dialect = getattr(statement_config, "dialect", None)
ddl_targets = (
(self._session_table, self._sessions_table_ddl()),
(self._events_table, self._events_table_ddl()),
(self._app_state_table, self._app_states_table_ddl()),
(self._user_state_table, self._user_states_table_ddl()),
(self._metadata_table, self._metadata_table_ddl()),
)
targets = [SchemaTarget.from_ddl(table, ddl, dialect=dialect) for table, ddl in ddl_targets]
_reconcile_adk_schema_sync(self._config, targets, create_schema=create_schema, assume_existing=assume_existing)
[docs]
def ensure_tables(self) -> None:
"""Create tables and emit a standardized log entry."""
if self.create_schema_enabled:
self.create_tables()
self.reconcile_schema(assume_existing=self.create_schema_enabled)
self._log_tables_created()
[docs]
def drop_tables(self) -> None:
"""Drop all ADK tables managed by this store in FK-safe order."""
self._execute_lifecycle_scripts(self._drop_tables_sql())
self._log_tables_dropped()
[docs]
def recreate_tables(self) -> None:
"""Drop and recreate all ADK tables managed by this store."""
self.drop_tables()
self.ensure_tables()
self._log_tables_recreated()
def _execute_lifecycle_scripts(self, statements: list[str]) -> None:
"""Execute lifecycle DDL scripts using the sync driver session."""
_run_lifecycle_sync(self._config, statements)
@abstractmethod
def _sessions_table_ddl(self) -> str:
"""Get the CREATE TABLE SQL for the sessions table."""
raise NotImplementedError
@abstractmethod
def _events_table_ddl(self) -> str:
"""Get the CREATE TABLE SQL for the events table."""
raise NotImplementedError
@abstractmethod
def _app_states_table_ddl(self) -> str:
"""Get the CREATE TABLE SQL for the app-scoped state table."""
raise NotImplementedError
@abstractmethod
def _user_states_table_ddl(self) -> str:
"""Get the CREATE TABLE SQL for the user-scoped state table."""
raise NotImplementedError
@abstractmethod
def _metadata_table_ddl(self) -> str:
"""Get the CREATE TABLE SQL for the ADK internal metadata table."""
raise NotImplementedError
@abstractmethod
def _drop_app_states_table_sql(self) -> str:
"""Get the DROP TABLE SQL statement for the app-scoped state table."""
raise NotImplementedError
@abstractmethod
def _drop_user_states_table_sql(self) -> str:
"""Get the DROP TABLE SQL statement for the user-scoped state table."""
raise NotImplementedError
@abstractmethod
def _drop_metadata_table_sql(self) -> str:
"""Get the DROP TABLE SQL statement for the ADK internal metadata table."""
raise NotImplementedError
@abstractmethod
def _drop_tables_sql(self) -> "list[str]":
"""Get the DROP TABLE SQL statements for this database dialect."""
raise NotImplementedError
def _run_lifecycle_sync(config: Any, statements: "list[str]") -> None:
"""Execute lifecycle statements through a synchronous config session."""
with cast("Any", config.provide_session()) as driver:
for statement in statements:
driver.execute_script(statement)
commit = getattr(driver, "commit", None)
if callable(commit):
commit()
def _reconcile_adk_schema_sync(
config: Any, targets: "list[SchemaTarget]", *, create_schema: bool, assume_existing: bool
) -> None:
"""Run additive reconciliation for a synchronous ADK store."""
with cast("Any", config.provide_session()) as driver:
ensure_schema_sync(
driver, targets, manage_schema=True, create_schema=create_schema, assume_existing=assume_existing
)
