Source code for sqlspec.adapters.oracledb.data_dictionary

"""Oracle-specific data dictionary for metadata queries."""

from enum import Enum
from typing import TYPE_CHECKING, Any, ClassVar, cast

from mypy_extensions import mypyc_attr

from sqlspec.data_dictionary import (
    ColumnMetadata,
    DDLResult,
    ForeignKeyMetadata,
    IndexMetadata,
    MetadataCapability,
    MetadataCapabilityProfile,
    MetadataFidelity,
    MetadataResult,
    MetadataRisk,
    MetadataSource,
    MetadataSupport,
    ObjectIdentity,
    SystemMetadataCapability,
    SystemMetadataRequest,
    SystemMetadataResult,
    TableMetadata,
    VersionInfo,
    ensure_system_metadata_request,
    get_data_dictionary_loader,
    get_dialect_config,
    system_metadata_gated_result,
)
from sqlspec.data_dictionary.dialects.oracle import (
    extract_oracle_version_value,
    list_oracle_available_features,
    merge_oracle_table_lists,
    oracle_supports_json_blob,
    oracle_supports_native_json,
    oracle_supports_oson_blob,
    parse_oracle_compatible_major,
    parse_oracle_version_components,
    resolve_oracle_feature_flag,
    resolve_oracle_json_storage,
    resolve_oracle_json_type,
)
from sqlspec.driver import AsyncDataDictionaryBase, SyncDataDictionaryBase
from sqlspec.utils.logging import get_logger
from sqlspec.utils.text import normalize_identifier, quote_identifier

if TYPE_CHECKING:
    from collections.abc import Sequence

    from sqlspec.adapters.oracledb.driver import OracleAsyncDriver, OracleSyncDriver
    from sqlspec.data_dictionary._types import DialectConfig

__all__ = (
    "JSONStorageType",
    "OracleVersionCache",
    "OracleVersionInfo",
    "OracledbAsyncDataDictionary",
    "OracledbSyncDataDictionary",
    "resolve_oracle_connection_major",
    "storage_type_from_version",
)

logger = get_logger("sqlspec.adapters.oracledb.data_dictionary")

ORACLE_DEFAULT_CAPABILITY_DOMAINS = (
    "schemas",
    "objects",
    "tables",
    "columns",
    "constraints",
    "indexes",
    "views",
    "materialized_views",
    "sequences",
    "routines",
    "arguments",
    "source",
    "triggers",
    "comments",
    "grants",
    "dependencies",
    "partitions",
    "lob_storage",
    "ddl",
    "system",
    "scope:user",
    "scope:all",
    "scope:dba",
    "scope:cdb",
)
ORACLE_PARTIAL_VISIBILITY_WARNING = "ALL_* views can omit inaccessible objects."
ORACLE_PRIVILEGED_SCOPE_WARNING = "DBA_* and CDB_* views require explicit privileged opt-in."
ORACLE_DIAGNOSTICS_WARNING = (
    "Oracle diagnostics, AWR/ASH/ADDM, Statspack, and DBA_HIST views require explicit opt-in, "
    "privileges, and license acknowledgement."
)
ORACLE_DDL_WARNINGS = (
    "DBMS_METADATA output depends on caller privileges and session transform settings.",
    "Native DDL can expose source text, comments, grants, storage details, and security-sensitive metadata.",
)
ORACLE_SYSTEM_METADATA_DOMAINS = ("system", "diagnostics", "awr", "ash", "addm", "statspack")
ORACLE_DISABLED_SYSTEM_METADATA_DOMAINS = ("no_diagnostics", "disabled")
ORACLE_SYSTEM_REDACTION_FIELDS = ("query_text", "sql_text", "username", "user_name", "session_user")
_STORAGE_OPTION_FIELD_COUNT = 2


class OracleVersionInfo(VersionInfo):
    """Oracle database version information."""

    def __init__(
        self, major: int, minor: int = 0, patch: int = 0, compatible: "str | None" = None, is_autonomous: bool = False
    ) -> None:
        """Initialize Oracle version info.

        Args:
            major: Major version number.
            minor: Minor version number.
            patch: Patch version number.
            compatible: Compatible parameter value.
            is_autonomous: Whether this is an Autonomous Database.
        """
        super().__init__(major, minor, patch)
        self.compatible = compatible
        self.is_autonomous = is_autonomous
@property def compatible_major(self) -> "int | None": """Get major version from compatible parameter.""" return parse_oracle_compatible_major(self.compatible) def supports_native_json(self) -> bool: """Check if database supports native JSON data type.""" return oracle_supports_native_json(self.major, self.compatible_major)