Add schema introspection via OData endpoints to Opteryx SQLAlchemy

README.md

@@ -12,6 +12,7 @@ This project packages a SQLAlchemy dialect and lightweight DBAPI 2.0 adapter tha
 - Read-only analytics with transparent polling of query status and incremental result streaming.
 - Lightweight DBAPI implementation that maps Opteryx types to SQLAlchemy while surfacing DatabaseError/OperationalError semantics.
 - Compatible with SQLAlchemy 2.x usage patterns, including context-managed engines and `text` queries.
+- Schema introspection (`inspect(engine)`) — list schemas, tables, views, and columns (with types) without running SQL.
 - Work with pandas, dbt, or other tooling that understands SQLAlchemy engines.
 - **Comprehensive debug logging** for troubleshooting connection, authentication, and query execution issues.
 - Install from PyPI (`pip install opteryx-sqlalchemy`) or lock into editable mode for development.
@@ -129,10 +130,54 @@ Note: this requires pandas to be installed in your environment.
 
 ---
 
+## Schema Introspection 🔍
+
+Use SQLAlchemy's `inspect()` to browse schemas, tables, views, and columns without writing SQL:
+
+```python
+from sqlalchemy import create_engine, inspect
+
+engine = create_engine(
+    "opteryx://myusername:mytoken@opteryx.app:443/default?ssl=true"
+)
+
+with engine.connect() as conn:
+    insp = inspect(conn)
+
+    # List every schema (namespace) visible to this token
+    print(insp.get_schema_names())
+    # ['personal.bastian', 'public.examples', 'public.github', ...]
+
+    # List tables — scoped to a schema, or unscoped for full dotted names
+    print(insp.get_table_names(schema="public.examples"))
+    # ['planets', 'moons', 'users']
+    print(insp.get_table_names())
+    # ['public.examples.planets', 'public.examples.moons', ...]
+
+    # Views are listed separately from tables
+    print(insp.get_view_names(schema="public.examples"))
+
+    # Check whether a table exists
+    print(insp.has_table("public.examples.planets"))
+    # True
+
+    # Get column names and types
+    for column in insp.get_columns("planets", schema="public.examples"):
+        print(column["name"], column["type"], "nullable:", column["nullable"])
+    # id BIGINT nullable: False
+    # name VARCHAR nullable: True
+    # mass FLOAT nullable: True
+    # ...
+```
+
+Introspection is backed by Opteryx's OData metadata endpoints rather than SQL queries, so it doesn't cost a billed query execution. The first call that needs table schemas (`get_columns`, or `has_table`/`get_table_names` the first time) can take longer than a typical query — full metadata generation is a heavier server-side operation — but the result is cached for the lifetime of the connection, so repeated calls (e.g. reflecting many tables) don't re-pay that cost.
+
+---
+
 ## Behavior and Limitations ⚠️
 
 - Opteryx is primarily an analytics engine — the dialect treats the service as read-only. Transactional features are effectively no-ops.
-- Not all SQLAlchemy reflection/introspection features are available. Some schema introspection operations may return empty results or limited metadata.
+- Schema introspection (`has_table`, `get_table_names`, `get_view_names`, `get_schema_names`, `get_columns`) is supported. Opteryx has no primary keys, foreign keys, or indexes, so `get_pk_constraint`/`get_foreign_keys`/`get_indexes` always return empty.
 - The dialect maps Opteryx native types to SQLAlchemy types as best-effort but does not implement a complete type mapping for every possible backend type.
 - If execution fails or times out, the DBAPI will raise an appropriate exception (subclass of DatabaseError/OperationalError).
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "opteryx-sqlalchemy"
-version = "0.0.7"
+version = "0.0.8"
 description = "Opteryx SQLAlchemy Dialect"
 requires-python = ">=3.9"
 dependencies = [

sqlalchemy_dialect/dbapi.py

@@ -196,6 +196,7 @@ def __init__(self, connection: "Connection") -> None:
                     try:
                         auth_header = f"{token_type} {token}"
                         self._connection._session.headers["Authorization"] = auth_header
+                        self._connection._jwt_authenticated = True
                         logger.debug("Set Authorization header to: %s ...", auth_header[:50])
                     except Exception as e:
                         logger.warning("Failed to set Authorization header: %s", e)
@@ -575,6 +576,9 @@ def __init__(
         self._ssl = ssl
         self._timeout = timeout
         self._closed = False
+        self._jwt_authenticated = False
+        self._odata_service_document_cache: Optional[List[Dict[str, Any]]] = None
+        self._odata_metadata_cache: Optional[Dict[str, List[Tuple[str, str, bool]]]] = None
 
         # Build base URL
         scheme = "https" if ssl else "http"
@@ -622,6 +626,113 @@ def _data_base_url(self) -> str:
             return f"{scheme}://{data_host}"
         return f"{scheme}://{data_host}:{self._port}"
 
+    def _odata_base_url(self) -> str:
+        """Construct a base URL that targets the 'odata' subdomain for metadata requests."""
+        scheme = "https" if self._ssl else "http"
+        domain = self._normalize_domain(self._host)
+        if "." in domain and not domain.startswith("localhost"):
+            odata_host = f"odata.{domain}"
+        else:
+            odata_host = domain
+        if (self._ssl and self._port == 443) or (not self._ssl and self._port == 80):
+            return f"{scheme}://{odata_host}"
+        return f"{scheme}://{odata_host}:{self._port}"
+
+    def _ensure_authenticated(self) -> None:
+        """Trigger the client-credentials auth flow if it hasn't run yet.
+
+        Authentication normally happens as a side effect of creating a Cursor
+        (see Cursor.__init__). Introspection calls can run before any query
+        cursor exists, so they need to force it explicitly.
+        """
+        if not self._jwt_authenticated:
+            self.cursor()
+
+    # $metadata generation is observed to take ~30s server-side regardless of
+    # payload size (~50KB) — give it more headroom than a typical query timeout.
+    _ODATA_METADATA_TIMEOUT = 90.0
+
+    def get_odata_service_document(self) -> List[Dict[str, Any]]:
+        """Fetch the OData service document listing all EntitySets visible to this token.
+
+        Cached for the lifetime of this connection: SQLAlchemy reflection
+        calls this once per has_table/get_table_names/get_columns/etc, and
+        the underlying request is not cheap.
+
+        Returns:
+            List of entity descriptors, each with at least "name" (dotted, e.g.
+            "public.examples.planets"), "kind" ("EntitySet"), and "source"
+            ("Table" or "View"). Returns an empty list on failure.
+        """
+        if self._odata_service_document_cache is not None:
+            return self._odata_service_document_cache
+
+        self._check_closed()
+        self._ensure_authenticated()
+        url = urljoin(self._odata_base_url() + "/", "api/v4/")
+        try:
+            response = self._session.get(url, timeout=self._ODATA_METADATA_TIMEOUT)
+            response.raise_for_status()
+            entities = response.json().get("value", [])
+            self._odata_service_document_cache = entities
+            return entities
+        except requests.exceptions.RequestException as e:
+            logger.warning("Failed to fetch OData service document: %s", e)
+            return []
+        except ValueError as e:
+            logger.warning("Failed to parse OData service document: %s", e)
+            return []
+
+    def get_odata_metadata(self) -> Dict[str, List[Tuple[str, str, bool]]]:
+        """Fetch and parse the OData $metadata document.
+
+        Cached for the lifetime of this connection (see get_odata_service_document).
+
+        Returns:
+            Mapping of EntityType name (dots replaced with underscores, e.g.
+            "public_examples_planets") to a list of
+            (property_name, edm_type, nullable) tuples. Returns an empty dict
+            on failure.
+        """
+        if self._odata_metadata_cache is not None:
+            return self._odata_metadata_cache
+
+        self._check_closed()
+        self._ensure_authenticated()
+        url = urljoin(self._odata_base_url() + "/", "api/v4/$metadata")
+        try:
+            response = self._session.get(url, timeout=self._ODATA_METADATA_TIMEOUT)
+            response.raise_for_status()
+        except requests.exceptions.RequestException as e:
+            logger.warning("Failed to fetch OData metadata: %s", e)
+            return {}
+
+        import xml.etree.ElementTree as ET
+
+        edm_ns = "{http://docs.oasis-open.org/odata/ns/edm}"
+        entity_types: Dict[str, List[Tuple[str, str, bool]]] = {}
+        try:
+            root = ET.fromstring(response.content)
+        except ET.ParseError as e:
+            logger.warning("Failed to parse OData $metadata XML: %s", e)
+            return {}
+
+        for entity_type in root.iter(f"{edm_ns}EntityType"):
+            name = entity_type.get("Name")
+            if not name:
+                continue
+            properties = []
+            for prop in entity_type.findall(f"{edm_ns}Property"):
+                prop_name = prop.get("Name")
+                if not prop_name:
+                    continue
+                prop_type = prop.get("Type", "Edm.String")
+                nullable = prop.get("Nullable", "true").lower() == "true"
+                properties.append((prop_name, prop_type, nullable))
+            entity_types[name] = properties
+        self._odata_metadata_cache = entity_types
+        return entity_types
+
     def _check_closed(self) -> None:
         """Raise exception if connection is closed."""
         if self._closed:

sqlalchemy_dialect/dialect.py

@@ -31,6 +31,38 @@
 logger = logging.getLogger("sqlalchemy.dialects.opteryx")
 
 
+_EDM_TYPE_MAP = {
+    "Edm.String": sqltypes.String,
+    "Edm.Boolean": sqltypes.Boolean,
+    "Edm.Byte": sqltypes.SmallInteger,
+    "Edm.SByte": sqltypes.SmallInteger,
+    "Edm.Int16": sqltypes.SmallInteger,
+    "Edm.Int32": sqltypes.Integer,
+    "Edm.Int64": sqltypes.BigInteger,
+    "Edm.Single": sqltypes.Float,
+    "Edm.Double": sqltypes.Float,
+    "Edm.Decimal": sqltypes.Numeric,
+    "Edm.Date": sqltypes.Date,
+    "Edm.TimeOfDay": sqltypes.Time,
+    "Edm.DateTimeOffset": sqltypes.DateTime,
+    "Edm.Binary": sqltypes.LargeBinary,
+}
+
+
+def _edm_type_to_sqlalchemy(edm_type: str) -> Any:
+    """Map an OData Edm.* primitive type name to a SQLAlchemy type."""
+    return _EDM_TYPE_MAP.get(edm_type, sqltypes.String)
+
+
+def _dbapi_connection(connection: Any) -> Any:
+    """Get the underlying opteryx dbapi Connection from a SQLAlchemy Connection.
+
+    Reuses the connection's already-authenticated HTTP session rather than
+    opening a new one, so introspection doesn't repeat the auth handshake.
+    """
+    return connection.connection.dbapi_connection
+
+
 def _quote_identifier(identifier: str) -> str:
     """Safely quote a SQL identifier to prevent SQL injection.
 
@@ -224,6 +256,9 @@ def has_table(
     ) -> bool:
         """Check if a table exists.
 
+        Backed by the OData service document rather than a SQL query, so this
+        costs a metadata lookup instead of a billed query execution.
+
         Args:
             connection: SQLAlchemy connection
             table_name: Name of the table
@@ -232,24 +267,9 @@ def has_table(
         Returns:
             True if the table exists
         """
-        # Try to query the table with a limit of 0 to check existence
-        try:
-            # Safely quote identifiers to prevent SQL injection
-            quoted_table = _quote_identifier(table_name)
-            if schema:
-                quoted_schema = _quote_identifier(schema)
-                full_name = f"{quoted_schema}.{quoted_table}"
-            else:
-                full_name = quoted_table
-            logger.debug("Checking if table exists: %s", full_name)
-            result = connection.execute(f"SELECT * FROM {full_name} LIMIT 0")
-            result.close()
-            logger.debug("Table exists: %s", full_name)
-            return True
-        except (ValueError, Exception) as e:
-            # ValueError from invalid identifier, or database error
-            logger.debug("Table does not exist or error checking: %s - %s", table_name, e)
-            return False
+        full_name = f"{schema}.{table_name}" if schema else table_name
+        entities = _dbapi_connection(connection).get_odata_service_document()
+        return any(e.get("name") == full_name for e in entities)
 
     def get_columns(
         self,
@@ -258,11 +278,22 @@ def get_columns(
         schema: Optional[str] = None,
         **kw: Any,
     ) -> list:
-        """Get column information for a table.
-
-        Returns an empty list as Opteryx may not support full introspection.
-        """
-        return []
+        """Get column information for a table from the OData $metadata document."""
+        full_name = f"{schema}.{table_name}" if schema else table_name
+        entity_type_name = full_name.replace(".", "_")
+        metadata = _dbapi_connection(connection).get_odata_metadata()
+        properties = metadata.get(entity_type_name)
+        if not properties:
+            return []
+        return [
+            {
+                "name": prop_name,
+                "type": _edm_type_to_sqlalchemy(edm_type),
+                "nullable": nullable,
+                "default": None,
+            }
+            for prop_name, edm_type, nullable in properties
+        ]
 
     def get_pk_constraint(
         self,
@@ -303,35 +334,46 @@ def get_indexes(
         """
         return []
 
-    def get_table_names(self, connection: Any, schema: Optional[str] = None, **kw: Any) -> list:
-        """Get list of table names.
+    def _get_entity_names(self, connection: Any, schema: Optional[str], source: str) -> list:
+        """Shared helper: entity names of a given OData `source` ("Table" or "View").
 
-        Attempts to query Opteryx for available tables.
+        If `schema` is given, names are scoped to that dotted prefix and
+        returned with the prefix stripped; otherwise the full dotted name is
+        returned (Opteryx datasets are commonly addressed in dotted form,
+        e.g. "public.examples.planets").
         """
-        try:
-            result = connection.execute("SHOW TABLES")
-            tables = [row[0] for row in result.fetchall()]
-            result.close()
-            return tables
-        except Exception:
-            return []
+        entities = _dbapi_connection(connection).get_odata_service_document()
+        prefix = f"{schema}." if schema else None
+        names = []
+        for entity in entities:
+            if entity.get("kind") != "EntitySet" or entity.get("source") != source:
+                continue
+            full_name = entity.get("name", "")
+            if prefix:
+                if not full_name.startswith(prefix):
+                    continue
+                names.append(full_name[len(prefix) :])
+            else:
+                names.append(full_name)
+        return names
 
-    def get_view_names(self, connection: Any, schema: Optional[str] = None, **kw: Any) -> list:
-        """Get list of view names.
+    def get_table_names(self, connection: Any, schema: Optional[str] = None, **kw: Any) -> list:
+        """Get list of table names from the OData service document."""
+        return self._get_entity_names(connection, schema, "Table")
 
-        Opteryx may not distinguish between tables and views.
-        """
-        return []
+    def get_view_names(self, connection: Any, schema: Optional[str] = None, **kw: Any) -> list:
+        """Get list of view names from the OData service document."""
+        return self._get_entity_names(connection, schema, "View")
 
     def get_schema_names(self, connection: Any, **kw: Any) -> list:
-        """Get list of schema names."""
-        try:
-            result = connection.execute("SHOW SCHEMAS")
-            schemas = [row[0] for row in result.fetchall()]
-            result.close()
-            return schemas
-        except Exception:
-            return ["default"]
+        """Get list of schema names, derived from the dotted prefixes of known entities."""
+        entities = _dbapi_connection(connection).get_odata_service_document()
+        schemas = set()
+        for entity in entities:
+            full_name = entity.get("name", "")
+            if "." in full_name:
+                schemas.add(full_name.rsplit(".", 1)[0])
+        return sorted(schemas)
 
     def _get_server_version_info(self, connection: Any) -> Tuple[int, ...]:
         """Get server version information."""