"""Oracle ADK store for Google Agent Development Kit session/event storage."""
from decimal import Decimal
from typing import TYPE_CHECKING, Any, Final, Literal, NoReturn, cast
from typing_extensions import NotRequired, TypedDict
from sqlspec import SQL
from sqlspec.adapters.oracledb._storage import oracle_table_feature_report, validate_oracle_identifier
from sqlspec.adapters.oracledb._typing import DatabaseError as OracleDatabaseError
from sqlspec.adapters.oracledb.data_dictionary import JSONStorageType, OracleVersionInfo, storage_type_from_version
from sqlspec.config import ADKConfig
from sqlspec.extensions.adk import (
BaseAsyncADKStore,
BaseSyncADKStore,
StoredEvent,
StoredSession,
normalize_session_list_options,
)
from sqlspec.extensions.adk.memory.store import BaseAsyncADKMemoryStore, BaseSyncADKMemoryStore
from sqlspec.utils.logging import get_logger
from sqlspec.utils.serializers import from_json, to_json
from sqlspec.utils.type_guards import is_async_readable, is_readable
if TYPE_CHECKING:
from collections.abc import Sequence
from datetime import datetime, timedelta
from sqlspec.adapters.oracledb.config import OracleAsyncConfig, OracleSyncConfig
from sqlspec.extensions.adk import SessionOrderBy, StoredMemory
__all__ = (
"JSONStorageType",
"OracleADKCompressionConfig",
"OracleADKConfig",
"OracleADKPartitionConfig",
"OracleAsyncADKMemoryStore",
"OracleAsyncADKStore",
"OracleSyncADKMemoryStore",
"OracleSyncADKStore",
"coerce_decimal_values",
"storage_type_from_version",
)
logger = get_logger("sqlspec.adapters.oracledb.adk.store")
ORACLE_TABLE_NOT_FOUND_ERROR: Final = 942
ORACLE_DUPLICATE_KEY_ERROR: Final = 1
ORACLE_DEFAULT_SESSION_TABLE: Final = "adk_session"
ORACLE_DEFAULT_EVENTS_TABLE: Final = "adk_event"
ORACLE_DEFAULT_APP_STATE_TABLE: Final = "adk_app_state"
ORACLE_DEFAULT_USER_STATE_TABLE: Final = "adk_user_state"
ORACLE_DEFAULT_METADATA_TABLE: Final = "adk_internal_metadata"
_ADK_METADATA_TABLE_DDL_TEMPLATE = (
"\n"
" BEGIN\n"
" EXECUTE IMMEDIATE 'CREATE TABLE {0} (\n"
" key VARCHAR2(128) PRIMARY KEY,\n"
" value VARCHAR2(512) NOT NULL\n"
" )';\n"
" END;\n"
" "
)
_ADK_SESSIONS_TABLE_DDL_FOR_TYPE_TEMPLATE = ", {0}"
_ADK_SESSIONS_TABLE_DDL_FOR_TYPE_TEMPLATE_2 = (
"\n"
" BEGIN\n"
" EXECUTE IMMEDIATE 'CREATE TABLE {0} (\n"
" id VARCHAR2(128) PRIMARY KEY,\n"
" app_name VARCHAR2(128) NOT NULL,\n"
" user_id VARCHAR2(128) NOT NULL,\n"
" {1},\n"
" create_time TIMESTAMP WITH TIME ZONE DEFAULT SYSTIMESTAMP NOT NULL,\n"
" update_time TIMESTAMP WITH TIME ZONE DEFAULT SYSTIMESTAMP NOT NULL{2}\n"
" ){3}';\n"
" END;\n"
"\n"
" BEGIN\n"
" EXECUTE IMMEDIATE 'CREATE INDEX idx_{4}_app_user\n"
" ON {5}(app_name, user_id)';\n"
" END;\n"
"\n"
" BEGIN\n"
" EXECUTE IMMEDIATE 'CREATE INDEX idx_{6}_update_time\n"
" ON {7}(update_time DESC)';\n"
" END;\n"
" "
)
_ADK_EVENTS_TABLE_DDL_FOR_TYPE_TEMPLATE = (
"\n"
" BEGIN\n"
" EXECUTE IMMEDIATE 'CREATE TABLE {0} (\n"
" id VARCHAR2(128) PRIMARY KEY,\n"
" app_name VARCHAR2(128) NOT NULL,\n"
" user_id VARCHAR2(128) NOT NULL,\n"
" session_id VARCHAR2(128) NOT NULL,\n"
" invocation_id VARCHAR2(256),\n"
" timestamp TIMESTAMP WITH TIME ZONE DEFAULT SYSTIMESTAMP NOT NULL,\n"
" {1},\n"
" CONSTRAINT fk_{2}_session FOREIGN KEY (session_id)\n"
" REFERENCES {3}(id) ON DELETE CASCADE\n"
" ){4}';\n"
" END;\n"
"\n"
" BEGIN\n"
" EXECUTE IMMEDIATE 'CREATE INDEX idx_{5}_session\n"
" ON {6}(session_id, timestamp ASC)';\n"
" END;\n"
"\n"
" BEGIN\n"
" EXECUTE IMMEDIATE 'CREATE INDEX idx_{7}_invocation\n"
" ON {8}(invocation_id)';\n"
" END;\n"
"\n"
" BEGIN\n"
" EXECUTE IMMEDIATE 'CREATE INDEX idx_{9}_timestamp\n"
" ON {10}(timestamp ASC)';\n"
" END;\n"
"\n"
" BEGIN\n"
" EXECUTE IMMEDIATE 'CREATE INDEX idx_{11}_app_timestamp\n"
" ON {12}(app_name, timestamp ASC)';\n"
" END;\n"
" "
)
_ADK_APP_STATES_TABLE_DDL_FOR_TYPE_TEMPLATE = (
"\n"
" BEGIN\n"
" EXECUTE IMMEDIATE 'CREATE TABLE {0} (\n"
" app_name VARCHAR2(128) PRIMARY KEY,\n"
" {1},\n"
" update_time TIMESTAMP WITH TIME ZONE DEFAULT SYSTIMESTAMP NOT NULL\n"
" ){2}';\n"
" END;\n"
" "
)
_ADK_USER_STATES_TABLE_DDL_FOR_TYPE_TEMPLATE = (
"\n"
" BEGIN\n"
" EXECUTE IMMEDIATE 'CREATE TABLE {0} (\n"
" app_name VARCHAR2(128) NOT NULL,\n"
" user_id VARCHAR2(128) NOT NULL,\n"
" {1},\n"
" update_time TIMESTAMP WITH TIME ZONE DEFAULT SYSTIMESTAMP NOT NULL,\n"
" PRIMARY KEY (app_name, user_id)\n"
" ){2}';\n"
" END;\n"
" "
)
_ADK_MEMORY_TABLE_DDL_FOR_TYPE_TEMPLATE = ",\n {0}"
_ADK_MEMORY_TABLE_DDL_FOR_TYPE_TEMPLATE_2 = (
"\n"
" BEGIN\n"
" EXECUTE IMMEDIATE 'CREATE INDEX idx_{0}_fts\n"
" ON {1}(content_text) INDEXTYPE IS CTXSYS.CONTEXT';\n"
" END;\n"
" "
)
_ADK_MEMORY_TABLE_DDL_FOR_TYPE_TEMPLATE_3 = (
"\n"
" BEGIN\n"
" EXECUTE IMMEDIATE 'CREATE TABLE {0} (\n"
" id VARCHAR2(128) PRIMARY KEY,\n"
" session_id VARCHAR2(128) NOT NULL,\n"
" app_name VARCHAR2(128) NOT NULL,\n"
" user_id VARCHAR2(128) NOT NULL,\n"
" scope VARCHAR2(16) DEFAULT ''user'' NOT NULL,\n"
" event_id VARCHAR2(128) NOT NULL UNIQUE,\n"
" author VARCHAR2(256){1},\n"
" timestamp TIMESTAMP WITH TIME ZONE NOT NULL,\n"
" {2},\n"
" content_text CLOB NOT NULL,\n"
" inserted_at TIMESTAMP WITH TIME ZONE DEFAULT SYSTIMESTAMP NOT NULL\n"
" ){3}';\n"
" END;\n"
"\n"
" BEGIN\n"
" EXECUTE IMMEDIATE 'CREATE INDEX idx_{4}_app_scope_user_time\n"
" ON {5}(app_name, scope, user_id, timestamp DESC)';\n"
" END;\n"
"\n"
" BEGIN\n"
" EXECUTE IMMEDIATE 'CREATE INDEX idx_{6}_scope\n"
" ON {7}(app_name, scope)';\n"
" END;\n"
"\n"
" BEGIN\n"
" EXECUTE IMMEDIATE 'CREATE INDEX idx_{8}_session\n"
" ON {9}(session_id)';\n"
" END;\n"
" {10}\n"
" "
)
_ADK_JSON_COLUMN_DDL_TEMPLATE = "{0} JSON NOT NULL"
_ADK_JSON_COLUMN_DDL_TEMPLATE_2 = "{0} BLOB CHECK ({1} IS JSON) NOT NULL"
_ADK_JSON_COLUMN_DDL_TEMPLATE_3 = "{0} BLOB NOT NULL"
[docs]
class OracleADKCompressionConfig(TypedDict):
"""Oracle-specific ADK table compression settings."""
enabled: NotRequired[bool]
"""Enable Oracle table compression clauses."""
algorithm: NotRequired[str]
"""Oracle compression algorithm key."""
[docs]
class OracleADKPartitionConfig(TypedDict):
"""Oracle-specific ADK table partitioning settings."""
strategy: NotRequired[str]
"""Partitioning strategy, such as ``hash`` or ``range``."""
partition_count: NotRequired[int]
"""Hash partition count."""
partitions: NotRequired[int]
"""Alias for ``partition_count``."""
interval: NotRequired[str]
"""Range partition interval key."""
initial_less_than: NotRequired[str]
"""Initial range partition upper bound expression."""
partition_key: NotRequired[str]
"""Default partition key for all ADK tables."""
session_partition_key: NotRequired[str]
"""Partition key override for the session table."""
events_partition_key: NotRequired[str]
"""Partition key override for the events table."""
app_state_partition_key: NotRequired[str]
"""Partition key override for the app state table."""
user_state_partition_key: NotRequired[str]
"""Partition key override for the user state table."""
memory_partition_key: NotRequired[str]
"""Partition key override for the memory table."""
[docs]
class OracleADKConfig(ADKConfig):
"""Oracle ADK extension settings consumed by Oracle ADK stores."""
in_memory: NotRequired[bool]
"""Enable Oracle INMEMORY clauses on ADK tables."""
compression: NotRequired[OracleADKCompressionConfig]
"""Oracle ADK table compression settings."""
partitioning: NotRequired[OracleADKPartitionConfig]
"""Oracle ADK table partitioning settings."""
session_table_options: NotRequired[str]
"""Raw Oracle table options for the ADK session table."""
events_table_options: NotRequired[str]
"""Raw Oracle table options for the ADK events table."""
app_state_table_options: NotRequired[str]
"""Raw Oracle table options for the ADK app state table."""
user_state_table_options: NotRequired[str]
"""Raw Oracle table options for the ADK user state table."""
memory_table_options: NotRequired[str]
"""Raw Oracle table options for the ADK memory table."""
def coerce_decimal_values(value: Any) -> Any:
return _coerce_decimal_values(value)
class OracleAsyncADKStore(BaseAsyncADKStore["OracleAsyncConfig"]):
"""Oracle async ADK store using oracledb async driver.
Implements session and event storage for Google Agent Development Kit
using Oracle Database via the python-oracledb async driver. Provides:
- Session state management with version-specific JSON storage
- Full-fidelity event storage via ``event_data`` column
- Atomic ``append_event_and_update_state`` for durable session mutations
- TIMESTAMP WITH TIME ZONE for timezone-aware timestamps
- Foreign key constraints with cascade delete
- Efficient upserts using MERGE statement
Args:
config: OracleAsyncConfig with extension_config["adk"] settings.
Notes:
- JSON storage type detected by capability (native JSON, BLOB IS JSON, or plain LOB)
- event_data stored as JSON (21c+) or BLOB (older versions)
- TIMESTAMP WITH TIME ZONE for timezone-aware timestamps
- Named parameters using :param_name
- State merging handled at application level
- owner_id_column supports NUMBER, VARCHAR2, RAW for Oracle FK types
- Configuration is read from config.extension_config["adk"]
"""
__slots__ = ("_in_memory",)
def __init__(self, config: "OracleAsyncConfig") -> None:
"""Initialize Oracle ADK store.
Args:
config: OracleAsyncConfig instance.
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")
- owner_id_column: Optional owner FK column DDL (default: None)
- in_memory: Enable INMEMORY PRIORITY HIGH clause (default: False)
"""
super().__init__(config)
_configure_oracle_adk_session_tables(self, config)
adk_config = _adk_config(config)
self._in_memory: bool = bool(adk_config.get("in_memory", False))
async def create_tables(self) -> None:
"""Create the ADK tables that the data dictionary reports as missing.
Notes:
Detects Oracle version to determine optimal JSON storage type.
Consults ``data_dictionary.get_tables`` so existing tables are left
untouched instead of relying on an ORA-955 swallow.
"""
if not self.create_schema_enabled:
await self.reconcile_schema()
return
storage_type = await self._detect_json_storage_type()
logger.debug("Creating ADK tables with storage type: %s", storage_type)
async with self._config.provide_session() as driver:
existing = _existing_table_names(await driver.data_dictionary.get_tables(driver))
if _bare_table_name(self._session_table) not in existing:
await driver.execute_script(self._sessions_table_ddl_for_type(storage_type))
if _bare_table_name(self._events_table) not in existing:
await driver.execute_script(self._events_table_ddl_for_type(storage_type))
if _bare_table_name(self._app_state_table) not in existing:
await driver.execute_script(self._app_states_table_ddl_for_type(storage_type))
if _bare_table_name(self._user_state_table) not in existing:
await driver.execute_script(self._user_states_table_ddl_for_type(storage_type))
if _bare_table_name(self._metadata_table) not in existing:
await driver.execute_script(await self._metadata_table_ddl())
await driver.commit()
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 session identifier.
app_name: Application name.
user_id: User identifier.
state: Initial session state.
owner_id: Optional owner ID value for owner_id_column (if configured).
Returns:
Created session record.
Notes:
Uses SYSTIMESTAMP for create_time and update_time.
State is serialized using version-appropriate format.
owner_id is ignored if owner_id_column not configured.
"""
state_data = await self._serialize_state(state)
if self._owner_id_column_name:
sql = f"""
INSERT INTO {self._session_table} (id, app_name, user_id, state, create_time, update_time, {self._owner_id_column_name})
VALUES (:id, :app_name, :user_id, :state, SYSTIMESTAMP, SYSTIMESTAMP, :owner_id)
"""
params = {
"id": session_id,
"app_name": app_name,
"user_id": user_id,
"state": state_data,
"owner_id": owner_id,
}
else:
sql = f"""
INSERT INTO {self._session_table} (id, app_name, user_id, state, create_time, update_time)
VALUES (:id, :app_name, :user_id, :state, SYSTIMESTAMP, SYSTIMESTAMP)
"""
params = {"id": session_id, "app_name": app_name, "user_id": user_id, "state": state_data}
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(sql, params)
await conn.commit()
result = await self.get_session(app_name, user_id, session_id)
if result is None:
msg = "Failed to fetch created session"
raise RuntimeError(msg)
return result
async def get_session(
self, app_name: str, user_id: str, session_id: str, *, renew_for: "int | timedelta | None" = None
) -> "StoredSession | None":
"""Get session by ID.
Args:
app_name: Application name.
user_id: User identifier.
session_id: Session identifier.
renew_for: If positive, touch update_time while reading.
Returns:
Session record or None if not found.
Notes:
Oracle returns datetime objects for TIMESTAMP columns.
State is deserialized using version-appropriate format.
"""
try:
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
if renew_for is not None and self._calculate_expires_at(renew_for) is not None:
await cursor.execute(
f"UPDATE {self._session_table} SET update_time = SYSTIMESTAMP WHERE app_name = :app_name AND user_id = :user_id AND id = :id",
{"app_name": app_name, "user_id": user_id, "id": session_id},
)
await conn.commit()
await cursor.execute(
f"""
SELECT id, app_name, user_id, state, create_time, update_time
FROM {self._session_table}
WHERE app_name = :app_name AND user_id = :user_id AND id = :id
""",
{"app_name": app_name, "user_id": user_id, "id": session_id},
)
row = await cursor.fetchone()
if row is None:
return None
session_id_val, app_name, user_id, state_data, create_time, update_time = row
state = await self._deserialize_state(state_data)
return StoredSession(
id=session_id_val,
app_name=app_name,
user_id=user_id,
state=state,
create_time=create_time,
update_time=update_time,
)
except OracleDatabaseError as e:
error_obj = e.args[0] if e.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return None
raise
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: Application name.
user_id: User identifier.
session_id: Session identifier.
state: New state dictionary (replaces existing state).
Notes:
This replaces the entire state dictionary.
Updates update_time to current timestamp.
State is serialized using version-appropriate format.
"""
state_data = await self._serialize_state(state)
sql = f"""
UPDATE {self._session_table}
SET state = :state, update_time = SYSTIMESTAMP
WHERE app_name = :app_name AND user_id = :user_id AND id = :id
"""
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(
sql, {"state": state_data, "app_name": app_name, "user_id": user_id, "id": session_id}
)
await conn.commit()
async def list_sessions(
self,
app_name: str,
user_id: "str | None" = None,
*,
order_by: "SessionOrderBy" = "update_time",
descending: bool = True,
limit: "int | None" = None,
offset: "int | None" = None,
) -> "list[StoredSession]":
"""List sessions for an app, optionally filtered by user.
Args:
app_name: Application name.
user_id: User identifier. If None, lists all sessions for the app.
order_by: Timestamp column to sort on.
descending: Sort direction for the timestamp column and the id tie-break.
limit: Maximum number of sessions to return.
offset: Number of leading rows to skip.
Returns:
List of session records.
Notes:
Uses composite index on (app_name, user_id) when user_id is provided.
State is deserialized using version-appropriate format.
"""
column, direction, page_limit, page_offset = normalize_session_list_options(order_by, descending, limit, offset)
if page_limit == 0:
return []
sql, params = _session_list_query(
self._session_table, app_name, user_id, column, direction, page_limit, page_offset
)
try:
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(sql, params)
rows = await cursor.fetchall()
results = []
for row in rows:
state = await self._deserialize_state(row[3])
results.append(
StoredSession(
id=row[0],
app_name=row[1],
user_id=row[2],
state=state,
create_time=row[4],
update_time=row[5],
)
)
return results
except OracleDatabaseError as e:
error_obj = e.args[0] if e.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return []
raise
async def delete_session(self, app_name: str, user_id: str, session_id: str) -> None:
"""Delete session and all associated events (cascade).
Args:
app_name: Application name.
user_id: User identifier.
session_id: Session identifier.
Notes:
Foreign key constraint ensures events are cascade-deleted.
"""
sql = f"DELETE FROM {self._session_table} WHERE app_name = :app_name AND user_id = :user_id AND id = :id"
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(sql, {"app_name": app_name, "user_id": user_id, "id": session_id})
await conn.commit()
async def append_event(self, event_record: StoredEvent) -> None:
"""Append an event to a session.
Args:
event_record: Event record.
"""
sql = f"""
INSERT INTO {self._events_table} (
id, app_name, user_id, session_id, invocation_id, timestamp, event_data
) VALUES (
:id, :app_name, :user_id, :session_id, :invocation_id, :timestamp, :event_data
)
"""
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(
sql,
{
"id": event_record["id"],
"app_name": event_record["app_name"],
"user_id": event_record["user_id"],
"session_id": event_record["session_id"],
"invocation_id": event_record["invocation_id"],
"timestamp": event_record["timestamp"],
"event_data": await self._serialize_event_data(event_record["event_data"]),
},
)
await conn.commit()
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 session + scoped state.
All writes are executed within a single transaction so they succeed or
fail together.
"""
insert_sql = f"""
INSERT INTO {self._events_table} (
id, app_name, user_id, session_id, invocation_id, timestamp, event_data
) VALUES (
:id, :app_name, :user_id, :session_id, :invocation_id, :timestamp, :event_data
)
"""
state_data = await self._serialize_state(state)
update_sql = f"""
UPDATE {self._session_table}
SET state = :state, update_time = SYSTIMESTAMP
WHERE app_name = :app_name AND user_id = :user_id AND id = :id
"""
select_sql = f"""
SELECT id, app_name, user_id, state, create_time, update_time
FROM {self._session_table}
WHERE app_name = :app_name AND user_id = :user_id AND id = :id
"""
app_upsert_sql = f"""
MERGE INTO {self._app_state_table} target
USING (SELECT :app_name AS app_name, :state AS state FROM DUAL) source
ON (target.app_name = source.app_name)
WHEN MATCHED THEN
UPDATE SET target.state = source.state, target.update_time = SYSTIMESTAMP
WHEN NOT MATCHED THEN
INSERT (app_name, state, update_time)
VALUES (source.app_name, source.state, SYSTIMESTAMP)
"""
user_upsert_sql = f"""
MERGE INTO {self._user_state_table} target
USING (SELECT :app_name AS app_name, :user_id AS user_id, :state AS state FROM DUAL) source
ON (target.app_name = source.app_name AND target.user_id = source.user_id)
WHEN MATCHED THEN
UPDATE SET target.state = source.state, target.update_time = SYSTIMESTAMP
WHEN NOT MATCHED THEN
INSERT (app_name, user_id, state, update_time)
VALUES (source.app_name, source.user_id, source.state, SYSTIMESTAMP)
"""
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
try:
await cursor.execute(
update_sql, {"state": state_data, "app_name": app_name, "user_id": user_id, "id": session_id}
)
await cursor.execute(select_sql, {"app_name": app_name, "user_id": user_id, "id": session_id})
row = await cursor.fetchone()
if row is None:
_raise_session_not_found(session_id)
await cursor.execute(
insert_sql,
{
"id": event_record["id"],
"app_name": event_record["app_name"],
"user_id": event_record["user_id"],
"session_id": event_record["session_id"],
"invocation_id": event_record["invocation_id"],
"timestamp": event_record["timestamp"],
"event_data": await self._serialize_event_data(event_record["event_data"]),
},
)
if app_state is not None:
await cursor.execute(
app_upsert_sql, {"app_name": app_name, "state": await self._serialize_state(app_state)}
)
if user_state is not None:
await cursor.execute(
user_upsert_sql,
{
"app_name": app_name,
"user_id": user_id,
"state": await self._serialize_state(user_state),
},
)
await conn.commit()
except Exception:
await conn.rollback()
raise
session_id_val, row_app_name, row_user_id, state_data_row, create_time, update_time = row
return StoredSession(
id=session_id_val,
app_name=row_app_name,
user_id=row_user_id,
state=await self._deserialize_state(state_data_row),
create_time=create_time,
update_time=update_time,
)
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: Application name.
user_id: User identifier.
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 ASC.
"""
if limit == 0:
return []
where_clauses = ["s.app_name = :app_name", "s.user_id = :user_id", "e.session_id = :session_id"]
params: dict[str, Any] = {"app_name": app_name, "user_id": user_id, "session_id": session_id}
if after_timestamp is not None:
where_clauses.append("e.timestamp > :after_timestamp")
params["after_timestamp"] = after_timestamp
where_clause = " AND ".join(where_clauses)
limit_clause = ""
if limit is not None:
limit_clause = f" FETCH FIRST {limit} ROWS ONLY"
sql = f"""
SELECT e.id, e.session_id, e.invocation_id, e.timestamp, e.event_data, s.app_name, s.user_id
FROM {self._events_table} e
JOIN {self._session_table} s ON e.session_id = s.id
WHERE {where_clause}
ORDER BY e.timestamp ASC{limit_clause}
"""
try:
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(sql, params)
rows = await cursor.fetchall()
return [
StoredEvent(
id=row[0],
session_id=row[1],
invocation_id=_oracle_text_value(row[2]),
timestamp=row[3],
event_data=await self._deserialize_json_field(row[4]) or {},
app_name=row[5],
user_id=row[6],
)
for row in rows
]
except OracleDatabaseError as e:
error_obj = e.args[0] if e.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return []
raise
async def delete_expired_events(self, before: "datetime", app_name: "str | None" = None) -> int:
sql = f"DELETE FROM {self._events_table} WHERE timestamp < :before"
params: dict[str, Any] = {"before": before}
if app_name is not None:
sql += " AND app_name = :app_name"
params["app_name"] = app_name
try:
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(sql, params)
await conn.commit()
return cursor.rowcount if cursor.rowcount and cursor.rowcount > 0 else 0
except OracleDatabaseError as e:
error_obj = e.args[0] if e.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return 0
raise
async def delete_idle_sessions(self, updated_before: "datetime", app_name: "str | None" = None) -> int:
sql = f"DELETE FROM {self._session_table} WHERE update_time < :updated_before"
params: dict[str, Any] = {"updated_before": updated_before}
if app_name is not None:
sql += " AND app_name = :app_name"
params["app_name"] = app_name
try:
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(sql, params)
await conn.commit()
return cursor.rowcount if cursor.rowcount and cursor.rowcount > 0 else 0
except OracleDatabaseError as e:
error_obj = e.args[0] if e.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return 0
raise
async def delete_idle_user_states(self, updated_before: "datetime", app_name: "str | None" = None) -> int:
sql = f"DELETE FROM {self._user_state_table} WHERE update_time < :updated_before"
params: dict[str, Any] = {"updated_before": updated_before}
if app_name is not None:
sql += " AND app_name = :app_name"
params["app_name"] = app_name
try:
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(sql, params)
await conn.commit()
return cursor.rowcount if cursor.rowcount and cursor.rowcount > 0 else 0
except OracleDatabaseError as e:
error_obj = e.args[0] if e.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return 0
raise
async def get_app_state(self, app_name: str) -> "dict[str, Any] | None":
"""Return app-scoped state for an application."""
sql = f"SELECT state FROM {self._app_state_table} WHERE app_name = :app_name"
try:
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(sql, {"app_name": app_name})
row = await cursor.fetchone()
return await self._deserialize_state(row[0]) if row is not None else None
except OracleDatabaseError as e:
error_obj = e.args[0] if e.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return None
raise
async def get_user_state(self, app_name: str, user_id: str) -> "dict[str, Any] | None":
"""Return user-scoped state for an application user."""
sql = f"""
SELECT state
FROM {self._user_state_table}
WHERE app_name = :app_name AND user_id = :user_id
"""
try:
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(sql, {"app_name": app_name, "user_id": user_id})
row = await cursor.fetchone()
return await self._deserialize_state(row[0]) if row is not None else None
except OracleDatabaseError as e:
error_obj = e.args[0] if e.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return None
raise
async def upsert_app_state(self, app_name: str, state: "dict[str, Any]") -> None:
"""Insert or replace app-scoped state for an application."""
sql = f"""
MERGE INTO {self._app_state_table} target
USING (SELECT :app_name AS app_name, :state AS state FROM DUAL) source
ON (target.app_name = source.app_name)
WHEN MATCHED THEN
UPDATE SET target.state = source.state, target.update_time = SYSTIMESTAMP
WHEN NOT MATCHED THEN
INSERT (app_name, state, update_time)
VALUES (source.app_name, source.state, SYSTIMESTAMP)
"""
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(sql, {"app_name": app_name, "state": await self._serialize_state(state)})
await conn.commit()
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."""
sql = f"""
MERGE INTO {self._user_state_table} target
USING (SELECT :app_name AS app_name, :user_id AS user_id, :state AS state FROM DUAL) source
ON (target.app_name = source.app_name AND target.user_id = source.user_id)
WHEN MATCHED THEN
UPDATE SET target.state = source.state, target.update_time = SYSTIMESTAMP
WHEN NOT MATCHED THEN
INSERT (app_name, user_id, state, update_time)
VALUES (source.app_name, source.user_id, source.state, SYSTIMESTAMP)
"""
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(
sql, {"app_name": app_name, "user_id": user_id, "state": await self._serialize_state(state)}
)
await conn.commit()
async def get_metadata(self, key: str) -> "str | None":
"""Return a value from the ADK internal metadata table."""
sql = f"SELECT value FROM {self._metadata_table} WHERE key = :key"
try:
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(sql, {"key": key})
row = await cursor.fetchone()
return str(row[0]) if row is not None else None
except OracleDatabaseError as e:
error_obj = e.args[0] if e.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return None
raise
async def set_metadata(self, key: str, value: str) -> None:
"""Set a value in the ADK internal metadata table."""
sql = f"""
MERGE INTO {self._metadata_table} target
USING (SELECT :key AS key, :value AS value FROM DUAL) source
ON (target.key = source.key)
WHEN MATCHED THEN
UPDATE SET target.value = source.value
WHEN NOT MATCHED THEN
INSERT (key, value)
VALUES (source.key, source.value)
"""
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(sql, {"key": key, "value": value})
await conn.commit()
async def _sessions_table_ddl(self) -> str:
"""Get Oracle CREATE TABLE SQL for sessions table.
Auto-detects optimal JSON storage type based on Oracle version.
Result is cached to minimize database queries.
"""
storage_type = await self._detect_json_storage_type()
return self._sessions_table_ddl_for_type(storage_type)
async def _events_table_ddl(self) -> str:
"""Get Oracle CREATE TABLE SQL for events table.
Auto-detects optimal JSON storage type based on Oracle version.
Result is cached to minimize database queries.
"""
storage_type = await self._detect_json_storage_type()
return self._events_table_ddl_for_type(storage_type)
async def _app_states_table_ddl(self) -> str:
"""Get Oracle CREATE TABLE SQL for app-scoped state."""
storage_type = await self._detect_json_storage_type()
return self._app_states_table_ddl_for_type(storage_type)
async def _user_states_table_ddl(self) -> str:
"""Get Oracle CREATE TABLE SQL for user-scoped state."""
storage_type = await self._detect_json_storage_type()
return self._user_states_table_ddl_for_type(storage_type)
async def _metadata_table_ddl(self) -> str:
"""Get Oracle CREATE TABLE SQL for ADK internal metadata."""
return _ADK_METADATA_TABLE_DDL_TEMPLATE.format(self._metadata_table)
async def _detect_json_storage_type(self) -> JSONStorageType:
"""Resolve the JSON storage type from the pool-scoped Oracle version.
- Oracle 21c+ with compatible >= 20: native JSON type
- Oracle 12c+: BLOB with IS JSON constraint
- Oracle 11g and earlier: plain BLOB
"""
return storage_type_from_version(await self._get_version_info())
async def _get_version_info(self) -> "OracleVersionInfo | None":
"""Return the pool-scoped Oracle version through the data dictionary."""
cache = self._config._oracle_version_cache
if cache.resolved:
return cache.version
async with self._config.provide_session() as driver:
version_info = await driver.data_dictionary.get_version(driver)
if version_info is None:
logger.warning("Could not detect Oracle version, defaulting to BLOB_JSON storage")
return version_info
async def _serialize_state(self, state: "dict[str, Any]") -> "str | bytes":
"""Serialize state dictionary to appropriate format based on storage type.
Args:
state: State dictionary to serialize.
Returns:
JSON string for JSON_NATIVE, bytes for BLOB types.
"""
storage_type = await self._detect_json_storage_type()
if storage_type == JSONStorageType.JSON_NATIVE:
return to_json(state)
return to_json(state, as_bytes=True)
async def _deserialize_state(self, data: Any) -> "dict[str, Any]":
"""Deserialize state data from database format.
Args:
data: Data from database (may be LOB, str, bytes, or dict).
Returns:
Deserialized state dictionary.
Notes:
Handles LOB reading if data has read() method.
Oracle JSON type may return dict directly.
"""
if is_async_readable(data):
data = await data.read()
elif is_readable(data):
data = data.read()
if isinstance(data, dict):
return cast("dict[str, Any]", _coerce_decimal_values(data))
if isinstance(data, bytes):
return cast("dict[str, Any]", from_json(data))
if isinstance(data, str):
return cast("dict[str, Any]", from_json(data))
return cast("dict[str, Any]", from_json(str(data)))
async def _deserialize_json_field(self, data: Any) -> "dict[str, Any] | None":
"""Deserialize JSON payloads from Oracle JSON/BLOB/LOB values."""
if data is None:
return None
return await self._deserialize_state(data)
async def _serialize_event_data(self, event_data: Any) -> "str | bytes":
"""Serialize event_data to the configured Oracle JSON storage format."""
storage_type = await self._detect_json_storage_type()
event_data = _normalize_event_data_for_storage(event_data)
if storage_type == JSONStorageType.JSON_NATIVE:
return to_json(event_data)
return to_json(event_data, as_bytes=True)
async def _read_event_data(self, data: Any) -> str:
"""Read event_data from database, handling LOB types.
Args:
data: Data from database (may be LOB, str, or dict).
Returns:
JSON string.
"""
if is_async_readable(data):
data = await data.read()
elif is_readable(data):
data = data.read()
if isinstance(data, dict):
return to_json(data)
if isinstance(data, bytes):
return data.decode("utf-8")
return str(data)
def _sessions_table_ddl_for_type(self, storage_type: JSONStorageType) -> str:
"""Get Oracle CREATE TABLE SQL for sessions with specified storage type.
Args:
storage_type: JSON storage type to use.
Returns:
SQL statement to create adk_session table.
"""
if storage_type == JSONStorageType.JSON_NATIVE:
state_column = "state JSON NOT NULL"
elif storage_type == JSONStorageType.BLOB_JSON:
state_column = "state BLOB CHECK (state IS JSON) NOT NULL"
else:
state_column = "state BLOB NOT NULL"
owner_id_column_sql = (
_ADK_SESSIONS_TABLE_DDL_FOR_TYPE_TEMPLATE.format(self._owner_id_column_ddl)
if self._owner_id_column_ddl
else ""
)
table_clauses = _adk_table_feature_clause(
self._config,
"session",
in_memory=self._in_memory,
hash_partition_key="id",
range_partition_key="create_time",
)
return _ADK_SESSIONS_TABLE_DDL_FOR_TYPE_TEMPLATE_2.format(
self._session_table,
state_column,
owner_id_column_sql,
table_clauses,
self._session_table,
self._session_table,
self._session_table,
self._session_table,
)
def _events_table_ddl_for_type(self, storage_type: JSONStorageType) -> str:
"""Get Oracle CREATE TABLE SQL for events with specified storage type.
The events table stores the full ADK Event in ``event_data`` and
keeps scalar event fields indexed for efficient scoped reads.
Args:
storage_type: JSON storage type to use.
Returns:
SQL statement to create adk_event table.
"""
event_data_col = _event_data_column_ddl(storage_type)
table_clauses = _adk_table_feature_clause(
self._config,
"events",
in_memory=self._in_memory,
hash_partition_key="session_id",
range_partition_key="timestamp",
)
return _ADK_EVENTS_TABLE_DDL_FOR_TYPE_TEMPLATE.format(
self._events_table,
event_data_col,
self._events_table,
self._session_table,
table_clauses,
self._events_table,
self._events_table,
self._events_table,
self._events_table,
self._events_table,
self._events_table,
self._events_table,
self._events_table,
)
def _app_states_table_ddl_for_type(self, storage_type: JSONStorageType) -> str:
"""Get Oracle CREATE TABLE SQL for app-scoped state with specified storage type."""
state_column = _json_column_ddl("state", storage_type)
table_clauses = _adk_table_feature_clause(
self._config,
"app_state",
in_memory=self._in_memory,
hash_partition_key="app_name",
range_partition_key="update_time",
)
return _ADK_APP_STATES_TABLE_DDL_FOR_TYPE_TEMPLATE.format(self._app_state_table, state_column, table_clauses)
def _user_states_table_ddl_for_type(self, storage_type: JSONStorageType) -> str:
"""Get Oracle CREATE TABLE SQL for user-scoped state with specified storage type."""
state_column = _json_column_ddl("state", storage_type)
table_clauses = _adk_table_feature_clause(
self._config,
"user_state",
in_memory=self._in_memory,
hash_partition_key="user_id",
range_partition_key="update_time",
)
return _ADK_USER_STATES_TABLE_DDL_FOR_TYPE_TEMPLATE.format(self._user_state_table, state_column, table_clauses)
def _drop_app_states_table_sql(self) -> str:
return f"""
BEGIN
EXECUTE IMMEDIATE 'DROP TABLE {self._app_state_table}';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -942 THEN
RAISE;
END IF;
END;
"""
def _drop_user_states_table_sql(self) -> str:
return f"""
BEGIN
EXECUTE IMMEDIATE 'DROP TABLE {self._user_state_table}';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -942 THEN
RAISE;
END IF;
END;
"""
def _drop_metadata_table_sql(self) -> str:
return f"""
BEGIN
EXECUTE IMMEDIATE 'DROP TABLE {self._metadata_table}';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -942 THEN
RAISE;
END IF;
END;
"""
def _drop_tables_sql(self) -> "list[str]":
"""Get Oracle DROP TABLE SQL statements.
Returns:
List of SQL statements to drop tables and indexes.
Notes:
Order matters: drop events table (child) before sessions (parent).
Oracle automatically drops indexes when dropping tables.
"""
return [
self._drop_metadata_table_sql(),
self._drop_user_states_table_sql(),
self._drop_app_states_table_sql(),
f"""
BEGIN
EXECUTE IMMEDIATE 'DROP INDEX idx_{self._events_table}_session';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -1418 THEN
RAISE;
END IF;
END;
""",
f"""
BEGIN
EXECUTE IMMEDIATE 'DROP INDEX idx_{self._session_table}_update_time';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -1418 THEN
RAISE;
END IF;
END;
""",
f"""
BEGIN
EXECUTE IMMEDIATE 'DROP INDEX idx_{self._session_table}_app_user';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -1418 THEN
RAISE;
END IF;
END;
""",
f"""
BEGIN
EXECUTE IMMEDIATE 'DROP TABLE {self._events_table}';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -942 THEN
RAISE;
END IF;
END;
""",
f"""
BEGIN
EXECUTE IMMEDIATE 'DROP TABLE {self._session_table}';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -942 THEN
RAISE;
END IF;
END;
""",
]
class OracleSyncADKStore(BaseSyncADKStore["OracleSyncConfig"]):
"""Oracle synchronous ADK store using oracledb sync driver.
Implements session and event storage for Google Agent Development Kit
using Oracle Database via the python-oracledb synchronous driver. Provides:
- Session state management with version-specific JSON storage
- Full-fidelity event storage via ``event_data`` column
- Atomic ``create_event_and_update_state`` for durable session mutations
- TIMESTAMP WITH TIME ZONE for timezone-aware timestamps
- Foreign key constraints with cascade delete
- Efficient upserts using MERGE statement
Args:
config: OracleSyncConfig with extension_config["adk"] settings.
Notes:
- JSON storage type detected by capability (native JSON, BLOB IS JSON, or plain LOB)
- event_data stored as JSON (21c+) or BLOB (older versions)
- TIMESTAMP WITH TIME ZONE for timezone-aware timestamps
- Named parameters using :param_name
- State merging handled at application level
- owner_id_column supports NUMBER, VARCHAR2, RAW for Oracle FK types
- Configuration is read from config.extension_config["adk"]
"""
__slots__ = ("_in_memory",)
def __init__(self, config: "OracleSyncConfig") -> None:
"""Initialize Oracle synchronous ADK store.
Args:
config: OracleSyncConfig instance.
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")
- owner_id_column: Optional owner FK column DDL (default: None)
- in_memory: Enable INMEMORY PRIORITY HIGH clause (default: False)
"""
super().__init__(config)
_configure_oracle_adk_session_tables(self, config)
adk_config = _adk_config(config)
self._in_memory: bool = bool(adk_config.get("in_memory", False))
def create_tables(self) -> None:
"""Create the ADK tables that the data dictionary reports as missing.
Notes:
Detects Oracle version to determine optimal JSON storage type.
Consults ``data_dictionary.get_tables`` so existing tables are left
untouched instead of relying on an ORA-955 swallow.
"""
if not self.create_schema_enabled:
self.reconcile_schema()
return
storage_type = self._detect_json_storage_type()
logger.info("Creating ADK tables with storage type: %s", storage_type)
with self._config.provide_session() as driver:
existing = _existing_table_names(driver.data_dictionary.get_tables(driver))
if _bare_table_name(self._session_table) not in existing:
driver.execute_script(SQL(self._sessions_table_ddl_for_type(storage_type)))
if _bare_table_name(self._events_table) not in existing:
driver.execute_script(SQL(self._events_table_ddl_for_type(storage_type)))
if _bare_table_name(self._app_state_table) not in existing:
driver.execute_script(SQL(self._app_states_table_ddl_for_type(storage_type)))
if _bare_table_name(self._user_state_table) not in existing:
driver.execute_script(SQL(self._user_states_table_ddl_for_type(storage_type)))
if _bare_table_name(self._metadata_table) not in existing:
driver.execute_script(SQL(self._metadata_table_ddl()))
driver.commit()
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 session identifier.
app_name: Application name.
user_id: User identifier.
state: Initial session state.
owner_id: Optional owner ID value for owner_id_column (if configured).
Returns:
Created session record.
Notes:
Uses SYSTIMESTAMP for create_time and update_time.
State is serialized using version-appropriate format.
owner_id is ignored if owner_id_column not configured.
"""
state_data = self._serialize_state(state)
if self._owner_id_column_name:
sql = f"""
INSERT INTO {self._session_table} (id, app_name, user_id, state, create_time, update_time, {self._owner_id_column_name})
VALUES (:id, :app_name, :user_id, :state, SYSTIMESTAMP, SYSTIMESTAMP, :owner_id)
"""
params = {
"id": session_id,
"app_name": app_name,
"user_id": user_id,
"state": state_data,
"owner_id": owner_id,
}
else:
sql = f"""
INSERT INTO {self._session_table} (id, app_name, user_id, state, create_time, update_time)
VALUES (:id, :app_name, :user_id, :state, SYSTIMESTAMP, SYSTIMESTAMP)
"""
params = {"id": session_id, "app_name": app_name, "user_id": user_id, "state": state_data}
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, params)
conn.commit()
result = self.get_session(app_name, user_id, session_id)
if result is None:
msg = "Failed to fetch created session"
raise RuntimeError(msg)
return result
def get_session(
self, app_name: str, user_id: str, session_id: str, *, renew_for: "int | timedelta | None" = None
) -> "StoredSession | None":
"""Get session by ID.
Args:
app_name: Application name.
user_id: User identifier.
session_id: Session identifier.
renew_for: If positive, touch update_time while reading.
Returns:
Session record or None if not found.
Notes:
Oracle returns datetime objects for TIMESTAMP columns.
State is deserialized using version-appropriate format.
"""
sql = f"""
SELECT id, app_name, user_id, state, create_time, update_time
FROM {self._session_table}
WHERE app_name = :app_name AND user_id = :user_id AND id = :id
"""
try:
with self._config.provide_connection() as conn, conn.cursor() as cursor:
if renew_for is not None and self._calculate_expires_at(renew_for) is not None:
cursor.execute(
f"UPDATE {self._session_table} SET update_time = SYSTIMESTAMP WHERE app_name = :app_name AND user_id = :user_id AND id = :id",
{"app_name": app_name, "user_id": user_id, "id": session_id},
)
conn.commit()
cursor.execute(sql, {"app_name": app_name, "user_id": user_id, "id": session_id})
row = cursor.fetchone()
if row is None:
return None
session_id_val, app_name, user_id, state_data, create_time, update_time = row
state = self._deserialize_state(state_data)
return StoredSession(
id=session_id_val,
app_name=app_name,
user_id=user_id,
state=state,
create_time=create_time,
update_time=update_time,
)
except OracleDatabaseError as e:
error_obj = e.args[0] if e.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return None
raise
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: Application name.
user_id: User identifier.
session_id: Session identifier.
state: New state dictionary (replaces existing state).
Notes:
This replaces the entire state dictionary.
Updates update_time to current timestamp.
State is serialized using version-appropriate format.
"""
state_data = self._serialize_state(state)
sql = f"""
UPDATE {self._session_table}
SET state = :state, update_time = SYSTIMESTAMP
WHERE app_name = :app_name AND user_id = :user_id AND id = :id
"""
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, {"state": state_data, "app_name": app_name, "user_id": user_id, "id": session_id})
conn.commit()
def list_sessions(
self,
app_name: str,
user_id: "str | None" = None,
*,
order_by: "SessionOrderBy" = "update_time",
descending: bool = True,
limit: "int | None" = None,
offset: "int | None" = None,
) -> "list[StoredSession]":
"""List sessions for an app, optionally filtered by user.
Args:
app_name: Application name.
user_id: User identifier. If None, lists all sessions for the app.
order_by: Timestamp column to sort on.
descending: Sort direction for the timestamp column and the id tie-break.
limit: Maximum number of sessions to return.
offset: Number of leading rows to skip.
Returns:
List of session records.
Notes:
Uses composite index on (app_name, user_id) when user_id is provided.
State is deserialized using version-appropriate format.
"""
column, direction, page_limit, page_offset = normalize_session_list_options(order_by, descending, limit, offset)
if page_limit == 0:
return []
sql, params = _session_list_query(
self._session_table, app_name, user_id, column, direction, page_limit, page_offset
)
try:
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, params)
rows = cursor.fetchall()
results = []
for row in rows:
state = self._deserialize_state(row[3])
results.append(
StoredSession(
id=row[0],
app_name=row[1],
user_id=row[2],
state=state,
create_time=row[4],
update_time=row[5],
)
)
return results
except OracleDatabaseError as e:
error_obj = e.args[0] if e.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return []
raise
def delete_session(self, app_name: str, user_id: str, session_id: str) -> None:
"""Delete session and all associated events (cascade).
Args:
app_name: Application name.
user_id: User identifier.
session_id: Session identifier.
Notes:
Foreign key constraint ensures events are cascade-deleted.
"""
sql = f"DELETE FROM {self._session_table} WHERE app_name = :app_name AND user_id = :user_id AND id = :id"
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, {"app_name": app_name, "user_id": user_id, "id": session_id})
conn.commit()
def append_event(self, event_record: StoredEvent) -> None:
"""Append an event to a session."""
sql = f"""
INSERT INTO {self._events_table} (
id, app_name, user_id, session_id, invocation_id, timestamp, event_data
) VALUES (
:id, :app_name, :user_id, :session_id, :invocation_id, :timestamp, :event_data
)
"""
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(
sql,
{
"id": event_record["id"],
"app_name": event_record["app_name"],
"user_id": event_record["user_id"],
"session_id": event_record["session_id"],
"invocation_id": event_record["invocation_id"],
"timestamp": event_record["timestamp"],
"event_data": self._serialize_event_data(event_record["event_data"]),
},
)
conn.commit()
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 session + scoped state."""
insert_sql = f"""
INSERT INTO {self._events_table} (
id, app_name, user_id, session_id, invocation_id, timestamp, event_data
) VALUES (
:id, :app_name, :user_id, :session_id, :invocation_id, :timestamp, :event_data
)
"""
state_data = self._serialize_state(state)
update_sql = f"""
UPDATE {self._session_table}
SET state = :state, update_time = SYSTIMESTAMP
WHERE app_name = :app_name AND user_id = :user_id AND id = :id
"""
select_sql = f"""
SELECT id, app_name, user_id, state, create_time, update_time
FROM {self._session_table}
WHERE app_name = :app_name AND user_id = :user_id AND id = :id
"""
app_upsert_sql = f"""
MERGE INTO {self._app_state_table} target
USING (SELECT :app_name AS app_name, :state AS state FROM DUAL) source
ON (target.app_name = source.app_name)
WHEN MATCHED THEN
UPDATE SET target.state = source.state, target.update_time = SYSTIMESTAMP
WHEN NOT MATCHED THEN
INSERT (app_name, state, update_time)
VALUES (source.app_name, source.state, SYSTIMESTAMP)
"""
user_upsert_sql = f"""
MERGE INTO {self._user_state_table} target
USING (SELECT :app_name AS app_name, :user_id AS user_id, :state AS state FROM DUAL) source
ON (target.app_name = source.app_name AND target.user_id = source.user_id)
WHEN MATCHED THEN
UPDATE SET target.state = source.state, target.update_time = SYSTIMESTAMP
WHEN NOT MATCHED THEN
INSERT (app_name, user_id, state, update_time)
VALUES (source.app_name, source.user_id, source.state, SYSTIMESTAMP)
"""
with self._config.provide_connection() as conn, conn.cursor() as cursor:
try:
cursor.execute(
update_sql, {"state": state_data, "app_name": app_name, "user_id": user_id, "id": session_id}
)
cursor.execute(select_sql, {"app_name": app_name, "user_id": user_id, "id": session_id})
row = cursor.fetchone()
if row is None:
_raise_session_not_found(session_id)
cursor.execute(
insert_sql,
{
"id": event_record["id"],
"app_name": event_record["app_name"],
"user_id": event_record["user_id"],
"session_id": event_record["session_id"],
"invocation_id": event_record["invocation_id"],
"timestamp": event_record["timestamp"],
"event_data": self._serialize_event_data(event_record["event_data"]),
},
)
if app_state is not None:
cursor.execute(app_upsert_sql, {"app_name": app_name, "state": self._serialize_state(app_state)})
if user_state is not None:
cursor.execute(
user_upsert_sql,
{"app_name": app_name, "user_id": user_id, "state": self._serialize_state(user_state)},
)
conn.commit()
except Exception:
conn.rollback()
raise
session_id_val, row_app_name, row_user_id, state_data_row, create_time, update_time = row
return StoredSession(
id=session_id_val,
app_name=row_app_name,
user_id=row_user_id,
state=self._deserialize_state(state_data_row),
create_time=create_time,
update_time=update_time,
)
def get_events(
self,
app_name: str,
user_id: str,
session_id: str,
after_timestamp: "datetime | None" = None,
limit: "int | None" = None,
) -> "list[StoredEvent]":
"""List events for a session ordered by timestamp.
Args:
app_name: Application name.
user_id: User identifier.
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 ASC.
"""
if limit == 0:
return []
where_clauses = ["s.app_name = :app_name", "s.user_id = :user_id", "e.session_id = :session_id"]
params: dict[str, Any] = {"app_name": app_name, "user_id": user_id, "session_id": session_id}
if after_timestamp is not None:
where_clauses.append("e.timestamp > :after_timestamp")
params["after_timestamp"] = after_timestamp
where_clause = " AND ".join(where_clauses)
limit_clause = f" FETCH FIRST {limit} ROWS ONLY" if limit is not None else ""
sql = f"""
SELECT e.id, e.session_id, e.invocation_id, e.timestamp, e.event_data, s.app_name, s.user_id
FROM {self._events_table} e
JOIN {self._session_table} s ON e.session_id = s.id
WHERE {where_clause}
ORDER BY e.timestamp ASC{limit_clause}
"""
try:
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, params)
rows = cursor.fetchall()
return [
StoredEvent(
id=row[0],
session_id=row[1],
invocation_id=_oracle_text_value(row[2]),
timestamp=row[3],
event_data=self._deserialize_json_field(row[4]) or {},
app_name=row[5],
user_id=row[6],
)
for row in rows
]
except OracleDatabaseError as e:
error_obj = e.args[0] if e.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return []
raise
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."""
sql = f"DELETE FROM {self._events_table} WHERE timestamp < :before"
params: dict[str, Any] = {"before": before}
if app_name is not None:
sql += " AND app_name = :app_name"
params["app_name"] = app_name
try:
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, params)
conn.commit()
return cursor.rowcount if cursor.rowcount and cursor.rowcount > 0 else 0
except OracleDatabaseError as e:
error_obj = e.args[0] if e.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return 0
raise
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."""
sql = f"DELETE FROM {self._session_table} WHERE update_time < :updated_before"
params: dict[str, Any] = {"updated_before": updated_before}
if app_name is not None:
sql += " AND app_name = :app_name"
params["app_name"] = app_name
try:
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, params)
conn.commit()
return cursor.rowcount if cursor.rowcount and cursor.rowcount > 0 else 0
except OracleDatabaseError as e:
error_obj = e.args[0] if e.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return 0
raise
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, optionally scoped to one application."""
sql = f"DELETE FROM {self._user_state_table} WHERE update_time < :updated_before"
params: dict[str, Any] = {"updated_before": updated_before}
if app_name is not None:
sql += " AND app_name = :app_name"
params["app_name"] = app_name
try:
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, params)
conn.commit()
return cursor.rowcount if cursor.rowcount and cursor.rowcount > 0 else 0
except OracleDatabaseError as e:
error_obj = e.args[0] if e.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return 0
raise
def get_app_state(self, app_name: str) -> "dict[str, Any] | None":
"""Return app-scoped state for an application."""
sql = f"SELECT state FROM {self._app_state_table} WHERE app_name = :app_name"
try:
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, {"app_name": app_name})
row = cursor.fetchone()
return self._deserialize_state(row[0]) if row is not None else None
except OracleDatabaseError as e:
error_obj = e.args[0] if e.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return None
raise
def get_user_state(self, app_name: str, user_id: str) -> "dict[str, Any] | None":
"""Return user-scoped state for an application user."""
sql = f"""
SELECT state
FROM {self._user_state_table}
WHERE app_name = :app_name AND user_id = :user_id
"""
try:
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, {"app_name": app_name, "user_id": user_id})
row = cursor.fetchone()
return self._deserialize_state(row[0]) if row is not None else None
except OracleDatabaseError as e:
error_obj = e.args[0] if e.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return None
raise
def upsert_app_state(self, app_name: str, state: "dict[str, Any]") -> None:
"""Insert or replace app-scoped state for an application."""
sql = f"""
MERGE INTO {self._app_state_table} target
USING (SELECT :app_name AS app_name, :state AS state FROM DUAL) source
ON (target.app_name = source.app_name)
WHEN MATCHED THEN
UPDATE SET target.state = source.state, target.update_time = SYSTIMESTAMP
WHEN NOT MATCHED THEN
INSERT (app_name, state, update_time)
VALUES (source.app_name, source.state, SYSTIMESTAMP)
"""
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, {"app_name": app_name, "state": self._serialize_state(state)})
conn.commit()
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."""
sql = f"""
MERGE INTO {self._user_state_table} target
USING (SELECT :app_name AS app_name, :user_id AS user_id, :state AS state FROM DUAL) source
ON (target.app_name = source.app_name AND target.user_id = source.user_id)
WHEN MATCHED THEN
UPDATE SET target.state = source.state, target.update_time = SYSTIMESTAMP
WHEN NOT MATCHED THEN
INSERT (app_name, user_id, state, update_time)
VALUES (source.app_name, source.user_id, source.state, SYSTIMESTAMP)
"""
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, {"app_name": app_name, "user_id": user_id, "state": self._serialize_state(state)})
conn.commit()
def get_metadata(self, key: str) -> "str | None":
"""Return a value from the ADK internal metadata table."""
sql = f"SELECT value FROM {self._metadata_table} WHERE key = :key"
try:
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, {"key": key})
row = cursor.fetchone()
return str(row[0]) if row is not None else None
except OracleDatabaseError as e:
error_obj = e.args[0] if e.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return None
raise
def set_metadata(self, key: str, value: str) -> None:
"""Set a value in the ADK internal metadata table."""
sql = f"""
MERGE INTO {self._metadata_table} target
USING (SELECT :key AS key, :value AS value FROM DUAL) source
ON (target.key = source.key)
WHEN MATCHED THEN
UPDATE SET target.value = source.value
WHEN NOT MATCHED THEN
INSERT (key, value)
VALUES (source.key, source.value)
"""
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, {"key": key, "value": value})
conn.commit()
def _sessions_table_ddl(self) -> str:
"""Get Oracle CREATE TABLE SQL for sessions table.
Auto-detects optimal JSON storage type based on Oracle version.
Result is cached to minimize database queries.
"""
storage_type = self._detect_json_storage_type()
return self._sessions_table_ddl_for_type(storage_type)
def _events_table_ddl(self) -> str:
"""Get Oracle CREATE TABLE SQL for events table.
Auto-detects optimal JSON storage type based on Oracle version.
Result is cached to minimize database queries.
"""
storage_type = self._detect_json_storage_type()
return self._events_table_ddl_for_type(storage_type)
def _app_states_table_ddl(self) -> str:
"""Get Oracle CREATE TABLE SQL for app-scoped state."""
storage_type = self._detect_json_storage_type()
return self._app_states_table_ddl_for_type(storage_type)
def _user_states_table_ddl(self) -> str:
"""Get Oracle CREATE TABLE SQL for user-scoped state."""
storage_type = self._detect_json_storage_type()
return self._user_states_table_ddl_for_type(storage_type)
def _metadata_table_ddl(self) -> str:
"""Get Oracle CREATE TABLE SQL for ADK internal metadata."""
return _ADK_METADATA_TABLE_DDL_TEMPLATE.format(self._metadata_table)
def _detect_json_storage_type(self) -> JSONStorageType:
"""Resolve the JSON storage type from the pool-scoped Oracle version.
- Oracle 21c+ with compatible >= 20: native JSON type
- Oracle 12c+: BLOB with IS JSON constraint
- Oracle 11g and earlier: plain BLOB
"""
return storage_type_from_version(self._get_version_info())
def _get_version_info(self) -> "OracleVersionInfo | None":
"""Return the pool-scoped Oracle version through the data dictionary."""
cache = self._config._oracle_version_cache
if cache.resolved:
return cache.version
with self._config.provide_session() as driver:
version_info = driver.data_dictionary.get_version(driver)
if version_info is None:
logger.warning("Could not detect Oracle version, defaulting to BLOB_JSON storage")
return version_info
def _serialize_state(self, state: "dict[str, Any]") -> "str | bytes":
"""Serialize state dictionary to appropriate format based on storage type.
Args:
state: State dictionary to serialize.
Returns:
JSON string for JSON_NATIVE, bytes for BLOB types.
"""
storage_type = self._detect_json_storage_type()
if storage_type == JSONStorageType.JSON_NATIVE:
return to_json(state)
return to_json(state, as_bytes=True)
def _deserialize_state(self, data: Any) -> "dict[str, Any]":
"""Deserialize state data from database format.
Args:
data: Data from database (may be LOB, str, bytes, or dict).
Returns:
Deserialized state dictionary.
Notes:
Handles LOB reading if data has read() method.
Oracle JSON type may return dict directly.
"""
if is_readable(data):
data = data.read()
if isinstance(data, dict):
return cast("dict[str, Any]", _coerce_decimal_values(data))
if isinstance(data, bytes):
return cast("dict[str, Any]", from_json(data))
if isinstance(data, str):
return cast("dict[str, Any]", from_json(data))
return cast("dict[str, Any]", from_json(str(data)))
def _deserialize_json_field(self, data: Any) -> "dict[str, Any] | None":
"""Deserialize JSON payloads from Oracle JSON/BLOB/LOB values."""
if data is None:
return None
return self._deserialize_state(data)
def _serialize_event_data(self, event_data: Any) -> "str | bytes":
"""Serialize event_data to the configured Oracle JSON storage format."""
storage_type = self._detect_json_storage_type()
event_data = _normalize_event_data_for_storage(event_data)
if storage_type == JSONStorageType.JSON_NATIVE:
return to_json(event_data)
return to_json(event_data, as_bytes=True)
def _read_event_data(self, data: Any) -> str:
"""Read event_data from database, handling LOB types.
Args:
data: Data from database (may be LOB, str, or dict).
Returns:
JSON string.
"""
if is_readable(data):
data = data.read()
if isinstance(data, dict):
return to_json(data)
if isinstance(data, bytes):
return data.decode("utf-8")
return str(data)
def _sessions_table_ddl_for_type(self, storage_type: JSONStorageType) -> str:
"""Get Oracle CREATE TABLE SQL for sessions with specified storage type.
Args:
storage_type: JSON storage type to use.
Returns:
SQL statement to create adk_session table.
"""
if storage_type == JSONStorageType.JSON_NATIVE:
state_column = "state JSON NOT NULL"
elif storage_type == JSONStorageType.BLOB_JSON:
state_column = "state BLOB CHECK (state IS JSON) NOT NULL"
else:
state_column = "state BLOB NOT NULL"
owner_id_column_sql = (
_ADK_SESSIONS_TABLE_DDL_FOR_TYPE_TEMPLATE.format(self._owner_id_column_ddl)
if self._owner_id_column_ddl
else ""
)
table_clauses = _adk_table_feature_clause(
self._config,
"session",
in_memory=self._in_memory,
hash_partition_key="id",
range_partition_key="create_time",
)
return _ADK_SESSIONS_TABLE_DDL_FOR_TYPE_TEMPLATE_2.format(
self._session_table,
state_column,
owner_id_column_sql,
table_clauses,
self._session_table,
self._session_table,
self._session_table,
self._session_table,
)
def _events_table_ddl_for_type(self, storage_type: JSONStorageType) -> str:
"""Get Oracle CREATE TABLE SQL for events with specified storage type.
The events table stores the full ADK Event in ``event_data`` and
keeps scalar event fields indexed for efficient scoped reads.
Args:
storage_type: JSON storage type to use.
Returns:
SQL statement to create adk_event table.
"""
event_data_col = _event_data_column_ddl(storage_type)
table_clauses = _adk_table_feature_clause(
self._config,
"events",
in_memory=self._in_memory,
hash_partition_key="session_id",
range_partition_key="timestamp",
)
return _ADK_EVENTS_TABLE_DDL_FOR_TYPE_TEMPLATE.format(
self._events_table,
event_data_col,
self._events_table,
self._session_table,
table_clauses,
self._events_table,
self._events_table,
self._events_table,
self._events_table,
self._events_table,
self._events_table,
self._events_table,
self._events_table,
)
def _app_states_table_ddl_for_type(self, storage_type: JSONStorageType) -> str:
"""Get Oracle CREATE TABLE SQL for app-scoped state with specified storage type."""
state_column = _json_column_ddl("state", storage_type)
table_clauses = _adk_table_feature_clause(
self._config,
"app_state",
in_memory=self._in_memory,
hash_partition_key="app_name",
range_partition_key="update_time",
)
return _ADK_APP_STATES_TABLE_DDL_FOR_TYPE_TEMPLATE.format(self._app_state_table, state_column, table_clauses)
def _user_states_table_ddl_for_type(self, storage_type: JSONStorageType) -> str:
"""Get Oracle CREATE TABLE SQL for user-scoped state with specified storage type."""
state_column = _json_column_ddl("state", storage_type)
table_clauses = _adk_table_feature_clause(
self._config,
"user_state",
in_memory=self._in_memory,
hash_partition_key="user_id",
range_partition_key="update_time",
)
return _ADK_USER_STATES_TABLE_DDL_FOR_TYPE_TEMPLATE.format(self._user_state_table, state_column, table_clauses)
def _drop_app_states_table_sql(self) -> str:
return f"""
BEGIN
EXECUTE IMMEDIATE 'DROP TABLE {self._app_state_table}';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -942 THEN
RAISE;
END IF;
END;
"""
def _drop_user_states_table_sql(self) -> str:
return f"""
BEGIN
EXECUTE IMMEDIATE 'DROP TABLE {self._user_state_table}';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -942 THEN
RAISE;
END IF;
END;
"""
def _drop_metadata_table_sql(self) -> str:
return f"""
BEGIN
EXECUTE IMMEDIATE 'DROP TABLE {self._metadata_table}';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -942 THEN
RAISE;
END IF;
END;
"""
def _drop_tables_sql(self) -> "list[str]":
"""Get Oracle DROP TABLE SQL statements.
Returns:
List of SQL statements to drop tables and indexes.
Notes:
Order matters: drop events table (child) before sessions (parent).
Oracle automatically drops indexes when dropping tables.
"""
return [
self._drop_metadata_table_sql(),
self._drop_user_states_table_sql(),
self._drop_app_states_table_sql(),
f"""
BEGIN
EXECUTE IMMEDIATE 'DROP INDEX idx_{self._events_table}_session';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -1418 THEN
RAISE;
END IF;
END;
""",
f"""
BEGIN
EXECUTE IMMEDIATE 'DROP INDEX idx_{self._session_table}_update_time';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -1418 THEN
RAISE;
END IF;
END;
""",
f"""
BEGIN
EXECUTE IMMEDIATE 'DROP INDEX idx_{self._session_table}_app_user';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -1418 THEN
RAISE;
END IF;
END;
""",
f"""
BEGIN
EXECUTE IMMEDIATE 'DROP TABLE {self._events_table}';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -942 THEN
RAISE;
END IF;
END;
""",
f"""
BEGIN
EXECUTE IMMEDIATE 'DROP TABLE {self._session_table}';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -942 THEN
RAISE;
END IF;
END;
""",
]
class OracleAsyncADKMemoryStore(BaseAsyncADKMemoryStore["OracleAsyncConfig"]):
"""Oracle ADK memory store using async oracledb driver."""
__slots__ = ("_in_memory",)
def __init__(self, config: "OracleAsyncConfig") -> None:
super().__init__(config)
adk_config = _adk_config(config)
self._in_memory: bool = bool(adk_config.get("in_memory", False))
async def create_tables(self) -> None:
if not self.create_schema_enabled:
await self.reconcile_schema()
return
if not self._enabled:
return
async with self._config.provide_session() as driver:
existing = _existing_table_names(await driver.data_dictionary.get_tables(driver))
if _bare_table_name(self._memory_table) not in existing:
await driver.execute_script(await self._memory_table_ddl())
async def insert_memory_entries(self, entries: "list[StoredMemory]", owner_id: "object | None" = None) -> int:
if not self._enabled:
msg = "Memory store is disabled"
raise RuntimeError(msg)
if not entries:
return 0
owner_column = f", {self._owner_id_column_name}" if self._owner_id_column_name else ""
owner_param = ", :owner_id" if self._owner_id_column_name else ""
sql = f"""
INSERT INTO {self._memory_table} (
id, session_id, app_name, user_id, scope, event_id, author{owner_column},
timestamp, content_json, content_text, metadata_json, inserted_at
) VALUES (
:id, :session_id, :app_name, :user_id, :scope, :event_id, :author{owner_param},
:timestamp, :content_json, :content_text, :metadata_json, :inserted_at
)
"""
inserted_count = 0
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
for entry in entries:
content_json = await self._serialize_json_field(entry["content_json"])
metadata_json = await self._serialize_json_field(entry["metadata_json"])
params = {
"id": entry["id"],
"session_id": entry["session_id"],
"app_name": entry["app_name"],
"user_id": entry["user_id"],
"scope": entry.get("scope", "user"),
"event_id": entry["event_id"],
"author": entry["author"],
"timestamp": entry["timestamp"],
"content_json": content_json,
"content_text": entry["content_text"],
"metadata_json": metadata_json,
"inserted_at": entry["inserted_at"],
}
if self._owner_id_column_name:
params["owner_id"] = str(owner_id) if owner_id is not None else None
if await self._execute_insert_entry(cursor, sql, params):
inserted_count += 1
await conn.commit()
return inserted_count
async def search_entries(
self,
query: str,
app_name: str,
user_id: str,
limit: "int | None" = None,
scope_filter: Literal["all", "user", "app"] = "all",
embedding: "Sequence[float] | None" = None,
) -> "list[StoredMemory]":
if not self._enabled:
msg = "Memory store is disabled"
raise RuntimeError(msg)
effective_limit = limit if limit is not None else self._max_results
try:
if self._use_fts:
return await self._search_entries_fts(query, app_name, user_id, effective_limit, scope_filter)
return await self._search_entries_simple(query, app_name, user_id, effective_limit, scope_filter)
except OracleDatabaseError as exc:
error_obj = exc.args[0] if exc.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return []
raise
async def delete_entries_by_session(self, session_id: str) -> int:
sql = f"DELETE FROM {self._memory_table} WHERE session_id = :session_id"
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(sql, {"session_id": session_id})
await conn.commit()
return cursor.rowcount if cursor.rowcount and cursor.rowcount > 0 else 0
async def delete_entries_older_than(
self, days: int, app_name: "str | None" = None, scope: "str | None" = None
) -> int:
clauses = ["inserted_at < SYSTIMESTAMP - NUMTODSINTERVAL(:days, 'DAY')"]
params: dict[str, Any] = {"days": days}
if app_name is not None:
clauses.append("app_name = :app_name")
params["app_name"] = app_name
if scope is not None:
clauses.append("scope = :scope")
params["scope"] = scope
where_sql = " AND ".join(clauses)
sql = f"DELETE FROM {self._memory_table} WHERE {where_sql}"
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(sql, params)
await conn.commit()
return cursor.rowcount if cursor.rowcount and cursor.rowcount > 0 else 0
async def _detect_json_storage_type(self) -> "JSONStorageType":
return storage_type_from_version(await self._get_version_info())
async def _get_version_info(self) -> "OracleVersionInfo | None":
cache = self._config._oracle_version_cache
if cache.resolved:
return cache.version
async with self._config.provide_session() as driver:
version_info = await driver.data_dictionary.get_version(driver)
if version_info is None:
logger.warning("Could not detect Oracle version, defaulting to BLOB_JSON storage")
return version_info
async def _serialize_json_field(self, value: Any) -> "str | bytes | None":
if value is None:
return None
storage_type = await self._detect_json_storage_type()
if storage_type == JSONStorageType.JSON_NATIVE:
return to_json(value)
return to_json(value, as_bytes=True)
async def _deserialize_json_field(self, data: Any) -> "dict[str, Any] | None":
if data is None:
return None
if is_async_readable(data) or is_readable(data):
data = await _read_lob_async(data)
return _extract_json_value(data)
async def _memory_table_ddl(self) -> str:
storage_type = await self._detect_json_storage_type()
return self._memory_table_ddl_for_type(storage_type)
def _memory_table_ddl_for_type(self, storage_type: "JSONStorageType") -> str:
if storage_type == JSONStorageType.JSON_NATIVE:
json_columns = """
content_json JSON,
metadata_json JSON
"""
elif storage_type == JSONStorageType.BLOB_JSON:
json_columns = """
content_json BLOB CHECK (content_json IS JSON),
metadata_json BLOB CHECK (metadata_json IS JSON)
"""
else:
json_columns = """
content_json BLOB,
metadata_json BLOB
"""
owner_id_line = (
_ADK_MEMORY_TABLE_DDL_FOR_TYPE_TEMPLATE.format(self._owner_id_column_ddl)
if self._owner_id_column_ddl
else ""
)
table_clauses = _adk_table_feature_clause(
self._config,
"memory",
in_memory=self._in_memory,
hash_partition_key="id",
range_partition_key="inserted_at",
)
fts_index = ""
if self._use_fts:
fts_index = _ADK_MEMORY_TABLE_DDL_FOR_TYPE_TEMPLATE_2.format(self._memory_table, self._memory_table)
return _ADK_MEMORY_TABLE_DDL_FOR_TYPE_TEMPLATE_3.format(
self._memory_table,
owner_id_line,
json_columns,
table_clauses,
self._memory_table,
self._memory_table,
self._memory_table,
self._memory_table,
self._memory_table,
self._memory_table,
fts_index,
)
def _drop_memory_table_sql(self) -> "list[str]":
return [
f"""
BEGIN
EXECUTE IMMEDIATE 'DROP INDEX idx_{self._memory_table}_session';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -1418 THEN
RAISE;
END IF;
END;
""",
f"""
BEGIN
EXECUTE IMMEDIATE 'DROP INDEX idx_{self._memory_table}_app_user_time';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -1418 THEN
RAISE;
END IF;
END;
""",
f"""
BEGIN
EXECUTE IMMEDIATE 'DROP TABLE {self._memory_table}';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -942 THEN
RAISE;
END IF;
END;
""",
]
async def _execute_insert_entry(self, cursor: Any, sql: str, params: "dict[str, Any]") -> bool:
"""Execute an insert and skip duplicate key errors."""
try:
await cursor.execute(sql, params)
except OracleDatabaseError as exc:
error_obj = exc.args[0] if exc.args else None
if error_obj and error_obj.code == ORACLE_DUPLICATE_KEY_ERROR:
return False
raise
return True
async def _search_entries_fts(
self, query: str, app_name: str, user_id: str, limit: int, scope_filter: Literal["all", "user", "app"] = "all"
) -> "list[StoredMemory]":
where_scope, scope_params = _build_oracle_scope_where(app_name, user_id, scope_filter)
sql = f"""
SELECT id, session_id, app_name, user_id, scope, event_id, author,
timestamp, content_json, content_text, metadata_json, inserted_at
FROM (
SELECT id, session_id, app_name, user_id, scope, event_id, author,
timestamp, content_json, content_text, metadata_json, inserted_at,
SCORE(1) AS score
FROM {self._memory_table}
WHERE {where_scope}
AND CONTAINS(content_text, :query, 1) > 0
ORDER BY score DESC, timestamp DESC
)
WHERE ROWNUM <= :limit
"""
params = {**scope_params, "query": query, "limit": limit}
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(sql, params)
rows = await cursor.fetchall()
return await self._rows_to_records(rows)
async def _search_entries_simple(
self, query: str, app_name: str, user_id: str, limit: int, scope_filter: Literal["all", "user", "app"] = "all"
) -> "list[StoredMemory]":
where_scope, scope_params = _build_oracle_scope_where(app_name, user_id, scope_filter)
sql = f"""
SELECT id, session_id, app_name, user_id, scope, event_id, author,
timestamp, content_json, content_text, metadata_json, inserted_at
FROM (
SELECT id, session_id, app_name, user_id, scope, event_id, author,
timestamp, content_json, content_text, metadata_json, inserted_at
FROM {self._memory_table}
WHERE {where_scope}
AND LOWER(content_text) LIKE :pattern
ORDER BY timestamp DESC
)
WHERE ROWNUM <= :limit
"""
pattern = f"%{query.lower()}%"
params = {**scope_params, "pattern": pattern, "limit": limit}
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(sql, params)
rows = await cursor.fetchall()
return await self._rows_to_records(rows)
async def _rows_to_records(self, rows: "list[Any]") -> "list[StoredMemory]":
records: list[StoredMemory] = []
for row in rows:
content_json = await self._deserialize_json_field(row[8]) if row[8] is not None else {}
metadata_json = await self._deserialize_json_field(row[10])
content_text = row[9]
if is_async_readable(content_text) or is_readable(content_text):
content_text = await _read_lob_async(content_text)
records.append({
"id": row[0],
"session_id": row[1],
"app_name": row[2],
"user_id": row[3],
"scope": row[4],
"event_id": row[5],
"author": row[6],
"timestamp": row[7],
"content_json": cast("dict[str, Any]", content_json),
"content_text": str(content_text),
"metadata_json": metadata_json,
"inserted_at": row[11],
"embedding": None,
})
return records
class OracleSyncADKMemoryStore(BaseSyncADKMemoryStore["OracleSyncConfig"]):
"""Oracle ADK memory store using sync oracledb driver."""
__slots__ = ("_in_memory",)
def __init__(self, config: "OracleSyncConfig") -> None:
super().__init__(config)
adk_config = _adk_config(config)
self._in_memory = bool(adk_config.get("in_memory", False))
def create_tables(self) -> None:
"""Create the memory table when the data dictionary reports it missing."""
if not self.create_schema_enabled:
self.reconcile_schema()
return
if not self._enabled:
return
with self._config.provide_session() as driver:
existing = _existing_table_names(driver.data_dictionary.get_tables(driver))
if _bare_table_name(self._memory_table) not in existing:
driver.execute_script(self._memory_table_ddl())
def insert_memory_entries(self, entries: "list[StoredMemory]", owner_id: "object | None" = None) -> int:
"""Bulk insert memory entries with deduplication."""
if not self._enabled:
msg = "Memory store is disabled"
raise RuntimeError(msg)
if not entries:
return 0
owner_column = f", {self._owner_id_column_name}" if self._owner_id_column_name else ""
owner_param = ", :owner_id" if self._owner_id_column_name else ""
sql = f"""
INSERT INTO {self._memory_table} (
id, session_id, app_name, user_id, scope, event_id, author{owner_column},
timestamp, content_json, content_text, metadata_json, inserted_at
) VALUES (
:id, :session_id, :app_name, :user_id, :scope, :event_id, :author{owner_param},
:timestamp, :content_json, :content_text, :metadata_json, :inserted_at
)
"""
inserted_count = 0
with self._config.provide_connection() as conn, conn.cursor() as cursor:
for entry in entries:
content_json = self._serialize_json_field(entry["content_json"])
metadata_json = self._serialize_json_field(entry["metadata_json"])
params = {
"id": entry["id"],
"session_id": entry["session_id"],
"app_name": entry["app_name"],
"user_id": entry["user_id"],
"scope": entry.get("scope", "user"),
"event_id": entry["event_id"],
"author": entry["author"],
"timestamp": entry["timestamp"],
"content_json": content_json,
"content_text": entry["content_text"],
"metadata_json": metadata_json,
"inserted_at": entry["inserted_at"],
}
if self._owner_id_column_name:
params["owner_id"] = str(owner_id) if owner_id is not None else None
if self._execute_insert_entry(cursor, sql, params):
inserted_count += 1
conn.commit()
return inserted_count
def search_entries(
self,
query: str,
app_name: str,
user_id: str,
limit: "int | None" = None,
scope_filter: Literal["all", "user", "app"] = "all",
embedding: "Sequence[float] | None" = None,
) -> "list[StoredMemory]":
"""Search memory entries by text query."""
if not self._enabled:
msg = "Memory store is disabled"
raise RuntimeError(msg)
effective_limit = limit if limit is not None else self._max_results
try:
if self._use_fts:
return self._search_entries_fts(query, app_name, user_id, effective_limit, scope_filter)
return self._search_entries_simple(query, app_name, user_id, effective_limit, scope_filter)
except OracleDatabaseError as exc:
error_obj = exc.args[0] if exc.args else None
if error_obj and error_obj.code == ORACLE_TABLE_NOT_FOUND_ERROR:
return []
raise
def delete_entries_by_session(self, session_id: str) -> int:
"""Delete all memory entries for a specific session."""
sql = f"DELETE FROM {self._memory_table} WHERE session_id = :session_id"
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, {"session_id": session_id})
conn.commit()
return cursor.rowcount if cursor.rowcount and cursor.rowcount > 0 else 0
def delete_entries_older_than(self, days: int, app_name: "str | None" = None, scope: "str | None" = None) -> int:
"""Delete memory entries older than specified days."""
clauses = ["inserted_at < SYSTIMESTAMP - NUMTODSINTERVAL(:days, 'DAY')"]
params: dict[str, Any] = {"days": days}
if app_name is not None:
clauses.append("app_name = :app_name")
params["app_name"] = app_name
if scope is not None:
clauses.append("scope = :scope")
params["scope"] = scope
where_sql = " AND ".join(clauses)
sql = f"DELETE FROM {self._memory_table} WHERE {where_sql}"
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, params)
conn.commit()
return cursor.rowcount if cursor.rowcount and cursor.rowcount > 0 else 0
def _detect_json_storage_type(self) -> "JSONStorageType":
return storage_type_from_version(self._get_version_info())
def _get_version_info(self) -> "OracleVersionInfo | None":
cache = self._config._oracle_version_cache
if cache.resolved:
return cache.version
with self._config.provide_session() as driver:
version_info = driver.data_dictionary.get_version(driver)
if version_info is None:
logger.warning("Could not detect Oracle version, defaulting to BLOB_JSON storage")
return version_info
def _serialize_json_field(self, value: Any) -> "str | bytes | None":
if value is None:
return None
storage_type = self._detect_json_storage_type()
if storage_type == JSONStorageType.JSON_NATIVE:
return to_json(value)
return to_json(value, as_bytes=True)
def _deserialize_json_field(self, data: Any) -> "dict[str, Any] | None":
if data is None:
return None
if is_readable(data):
data = _read_lob_sync(data)
return _extract_json_value(data)
def _memory_table_ddl(self) -> str:
storage_type = self._detect_json_storage_type()
return self._memory_table_ddl_for_type(storage_type)
def _memory_table_ddl_for_type(self, storage_type: "JSONStorageType") -> str:
if storage_type == JSONStorageType.JSON_NATIVE:
json_columns = """
content_json JSON,
metadata_json JSON
"""
elif storage_type == JSONStorageType.BLOB_JSON:
json_columns = """
content_json BLOB CHECK (content_json IS JSON),
metadata_json BLOB CHECK (metadata_json IS JSON)
"""
else:
json_columns = """
content_json BLOB,
metadata_json BLOB
"""
owner_id_line = (
_ADK_MEMORY_TABLE_DDL_FOR_TYPE_TEMPLATE.format(self._owner_id_column_ddl)
if self._owner_id_column_ddl
else ""
)
table_clauses = _adk_table_feature_clause(
self._config,
"memory",
in_memory=self._in_memory,
hash_partition_key="id",
range_partition_key="inserted_at",
)
fts_index = ""
if self._use_fts:
fts_index = _ADK_MEMORY_TABLE_DDL_FOR_TYPE_TEMPLATE_2.format(self._memory_table, self._memory_table)
return _ADK_MEMORY_TABLE_DDL_FOR_TYPE_TEMPLATE_3.format(
self._memory_table,
owner_id_line,
json_columns,
table_clauses,
self._memory_table,
self._memory_table,
self._memory_table,
self._memory_table,
self._memory_table,
self._memory_table,
fts_index,
)
def _drop_memory_table_sql(self) -> "list[str]":
return [
f"""
BEGIN
EXECUTE IMMEDIATE 'DROP INDEX idx_{self._memory_table}_session';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -1418 THEN
RAISE;
END IF;
END;
""",
f"""
BEGIN
EXECUTE IMMEDIATE 'DROP INDEX idx_{self._memory_table}_app_user_time';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -1418 THEN
RAISE;
END IF;
END;
""",
f"""
BEGIN
EXECUTE IMMEDIATE 'DROP TABLE {self._memory_table}';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -942 THEN
RAISE;
END IF;
END;
""",
]
def _execute_insert_entry(self, cursor: Any, sql: str, params: "dict[str, Any]") -> bool:
"""Execute an insert and skip duplicate key errors."""
try:
cursor.execute(sql, params)
except OracleDatabaseError as exc:
error_obj = exc.args[0] if exc.args else None
if error_obj and error_obj.code == ORACLE_DUPLICATE_KEY_ERROR:
return False
raise
return True
def _search_entries_fts(
self, query: str, app_name: str, user_id: str, limit: int, scope_filter: Literal["all", "user", "app"] = "all"
) -> "list[StoredMemory]":
where_scope, scope_params = _build_oracle_scope_where(app_name, user_id, scope_filter)
sql = f"""
SELECT id, session_id, app_name, user_id, scope, event_id, author,
timestamp, content_json, content_text, metadata_json, inserted_at
FROM (
SELECT id, session_id, app_name, user_id, scope, event_id, author,
timestamp, content_json, content_text, metadata_json, inserted_at,
SCORE(1) AS score
FROM {self._memory_table}
WHERE {where_scope}
AND CONTAINS(content_text, :query, 1) > 0
ORDER BY score DESC, timestamp DESC
)
WHERE ROWNUM <= :limit
"""
params = {**scope_params, "query": query, "limit": limit}
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, params)
rows = cursor.fetchall()
return self._rows_to_records(rows)
def _search_entries_simple(
self, query: str, app_name: str, user_id: str, limit: int, scope_filter: Literal["all", "user", "app"] = "all"
) -> "list[StoredMemory]":
where_scope, scope_params = _build_oracle_scope_where(app_name, user_id, scope_filter)
sql = f"""
SELECT id, session_id, app_name, user_id, scope, event_id, author,
timestamp, content_json, content_text, metadata_json, inserted_at
FROM (
SELECT id, session_id, app_name, user_id, scope, event_id, author,
timestamp, content_json, content_text, metadata_json, inserted_at
FROM {self._memory_table}
WHERE {where_scope}
AND LOWER(content_text) LIKE :pattern
ORDER BY timestamp DESC
)
WHERE ROWNUM <= :limit
"""
pattern = f"%{query.lower()}%"
params = {**scope_params, "pattern": pattern, "limit": limit}
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, params)
rows = cursor.fetchall()
return self._rows_to_records(rows)
def _rows_to_records(self, rows: "list[Any]") -> "list[StoredMemory]":
records: list[StoredMemory] = []
for row in rows:
content_json = self._deserialize_json_field(row[8]) if row[8] is not None else {}
metadata_json = self._deserialize_json_field(row[10])
content_text = row[9]
if is_readable(content_text):
content_text = _read_lob_sync(content_text)
records.append({
"id": row[0],
"session_id": row[1],
"app_name": row[2],
"user_id": row[3],
"scope": row[4],
"event_id": row[5],
"author": row[6],
"timestamp": row[7],
"content_json": cast("dict[str, Any]", content_json),
"content_text": str(content_text),
"metadata_json": metadata_json,
"inserted_at": row[11],
"embedding": None,
})
return records
def _configure_oracle_adk_session_tables(store: Any, config: Any) -> None:
"""Apply Oracle clean-break ADK table names independent of shared-base drift."""
adk_config = _adk_config(config)
table_names = {
"_session_table": str(adk_config.get("session_table") or ORACLE_DEFAULT_SESSION_TABLE),
"_events_table": str(adk_config.get("events_table") or ORACLE_DEFAULT_EVENTS_TABLE),
"_app_state_table": str(adk_config.get("app_state_table") or ORACLE_DEFAULT_APP_STATE_TABLE),
"_user_state_table": str(adk_config.get("user_state_table") or ORACLE_DEFAULT_USER_STATE_TABLE),
"_metadata_table": str(adk_config.get("metadata_table") or ORACLE_DEFAULT_METADATA_TABLE),
}
for attribute_name, table_name in table_names.items():
validate_oracle_identifier(table_name, "table name")
setattr(store, attribute_name, table_name)
def _normalize_event_data_for_storage(event_data: Any) -> Any:
"""Return event data without ADK 2.2-invalid durable ``actions: null``."""
if isinstance(event_data, dict) and event_data.get("actions") is None:
normalized = dict(event_data)
normalized.pop("actions", None)
return normalized
return event_data
def _raise_session_not_found(session_id: str) -> NoReturn:
msg = f"Session {session_id} not found during append_event_and_update_state."
raise ValueError(msg)
def _coerce_decimal_values(value: Any) -> Any:
if isinstance(value, Decimal):
return float(value)
if isinstance(value, dict):
return {key: _coerce_decimal_values(val) for key, val in value.items()}
if isinstance(value, list):
return [_coerce_decimal_values(item) for item in value]
if isinstance(value, tuple):
return tuple(_coerce_decimal_values(item) for item in value)
if isinstance(value, set):
return {_coerce_decimal_values(item) for item in value}
if isinstance(value, frozenset):
return frozenset(_coerce_decimal_values(item) for item in value)
return value
def _oracle_text_value(value: Any) -> str:
"""Normalize Oracle VARCHAR2 values back to Python strings.
Oracle stores empty strings as ``NULL``. The ADK event contract allows
empty strings for fields like ``invocation_id``, so reads coerce ``NULL``
back to ``""``.
"""
return "" if value is None else str(value)
def _extract_json_value(data: Any) -> "dict[str, Any]":
"""Extract and coerce a dictionary from a database JSON payload."""
if isinstance(data, dict):
return cast("dict[str, Any]", coerce_decimal_values(data))
if isinstance(data, bytes):
return cast("dict[str, Any]", from_json(data))
if isinstance(data, str):
return cast("dict[str, Any]", from_json(data))
return cast("dict[str, Any]", from_json(str(data)))
def _event_data_column_ddl(storage_type: JSONStorageType) -> str:
"""Return the DDL fragment for the event_data column."""
if storage_type == JSONStorageType.JSON_NATIVE:
return "event_data JSON NOT NULL"
if storage_type == JSONStorageType.BLOB_JSON:
return "event_data BLOB CHECK (event_data IS JSON) NOT NULL"
return "event_data BLOB NOT NULL"
def _json_column_ddl(column_name: str, storage_type: JSONStorageType) -> str:
"""Return an Oracle JSON column DDL fragment for the configured storage type."""
if storage_type == JSONStorageType.JSON_NATIVE:
return _ADK_JSON_COLUMN_DDL_TEMPLATE.format(column_name)
if storage_type == JSONStorageType.BLOB_JSON:
return _ADK_JSON_COLUMN_DDL_TEMPLATE_2.format(column_name, column_name)
return _ADK_JSON_COLUMN_DDL_TEMPLATE_3.format(column_name)
def _bare_table_name(name: str) -> str:
"""Return the case-folded, schema-stripped table name for membership checks."""
return name.rsplit(".", 1)[-1].casefold()
def _existing_table_names(rows: "list[Any]") -> "set[str]":
"""Collapse ``data_dictionary.get_tables`` rows into a case-folded name set."""
return {str(row.get("table_name", "")).rsplit(".", 1)[-1].casefold() for row in rows}
def _adk_config(config: Any) -> OracleADKConfig:
extension_config = getattr(config, "extension_config", {})
if not isinstance(extension_config, dict):
return {}
adk_config = extension_config.get("adk", {})
if isinstance(adk_config, dict):
return cast("OracleADKConfig", adk_config)
return {}
def _adk_table_feature_clause(
config: Any, table_kind: str, *, in_memory: bool, hash_partition_key: str, range_partition_key: str
) -> str:
report = oracle_table_feature_report(
config,
"adk",
_adk_config(config),
table_kind,
in_memory=in_memory,
hash_partition_key=hash_partition_key,
range_partition_key=range_partition_key,
)
return report["clause"]
async def _read_lob_async(data: Any) -> Any:
if is_async_readable(data):
return await data.read()
if is_readable(data):
return data.read()
return data
def _read_lob_sync(data: Any) -> Any:
if is_readable(data):
return data.read()
return data
def _build_oracle_scope_where(
app_name: str, user_id: str, scope_filter: Literal["all", "user", "app"]
) -> tuple[str, dict[str, Any]]:
if scope_filter == "all":
return "app_name = :app_name AND ((scope = 'user' AND user_id = :user_id) OR scope = 'app')", {
"app_name": app_name,
"user_id": user_id,
}
if scope_filter == "user":
return "app_name = :app_name AND scope = 'user' AND user_id = :user_id", {
"app_name": app_name,
"user_id": user_id,
}
return "app_name = :app_name AND scope = 'app'", {"app_name": app_name}
def _session_list_query(
session_table: str,
app_name: str,
user_id: "str | None",
column: str,
direction: str,
limit: "int | None",
offset: int,
) -> "tuple[str, dict[str, Any]]":
"""Return the bounded session-list query and its named binds."""
params: dict[str, Any] = {"app_name": app_name}
where_clause = "app_name = :app_name"
if user_id is not None:
params["user_id"] = user_id
where_clause = f"{where_clause} AND user_id = :user_id"
page_clause = ""
if limit is not None:
params["page_limit"] = limit
params["page_offset"] = offset
page_clause = "\n OFFSET :page_offset ROWS FETCH NEXT :page_limit ROWS ONLY"
sql = f"""
SELECT id, app_name, user_id, state, create_time, update_time
FROM {session_table}
WHERE {where_clause}
ORDER BY {column} {direction}, id {direction}{page_clause}
"""
return sql, params
