diff --git a/README.md b/README.md index a3a896c..b7dfea0 100644 --- a/README.md +++ b/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). diff --git a/pyproject.toml b/pyproject.toml index 3f6f371..63cb6cf 100644 --- a/pyproject.toml +++ b/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 = [ diff --git a/sqlalchemy_dialect/dbapi.py b/sqlalchemy_dialect/dbapi.py index b634cba..b0852c0 100644 --- a/sqlalchemy_dialect/dbapi.py +++ b/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: diff --git a/sqlalchemy_dialect/dialect.py b/sqlalchemy_dialect/dialect.py index 19351a1..65eb344 100644 --- a/sqlalchemy_dialect/dialect.py +++ b/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.""" diff --git a/tests/test_driver.py b/tests/test_driver.py index bbf97f9..a99b330 100644 --- a/tests/test_driver.py +++ b/tests/test_driver.py @@ -4,6 +4,7 @@ from unittest.mock import patch import pytest +import requests from sqlalchemy_dialect import dbapi from sqlalchemy_dialect.dialect import OpteryxDialect from sqlalchemy_dialect.dialect import _quote_identifier @@ -125,6 +126,141 @@ def test_rollback_noop(self): conn.rollback() # Should not raise conn.close() + def test_odata_base_url(self): + """Test that the odata subdomain is derived the same way as the jobs subdomain.""" + conn = dbapi.Connection(host="opteryx.app", port=443, ssl=True) + assert conn._odata_base_url() == "https://odata.opteryx.app" + conn.close() + + conn = dbapi.Connection(host="jobs.opteryx.app", port=443, ssl=True) + assert conn._odata_base_url() == "https://odata.opteryx.app" + conn.close() + + conn = dbapi.Connection(host="localhost", port=8000, ssl=False) + assert conn._odata_base_url() == "http://localhost:8000" + conn.close() + + @patch("requests.Session.get") + @patch("requests.Session.post") + def test_ensure_authenticated_triggers_jwt_flow(self, mock_post, mock_get): + """A pre-configured token header is not proof of a completed JWT exchange.""" + post_response = MagicMock() + post_response.status_code = 200 + post_response.json.return_value = {"access_token": "jwt-token", "token_type": "bearer"} + mock_post.return_value = post_response + + conn = dbapi.Connection(host="opteryx.app", username="user", token="raw-token", ssl=True) + # Seeded from the raw token before any JWT exchange has happened. + assert conn._session.headers["Authorization"] == "Bearer raw-token" + assert conn._jwt_authenticated is False + + conn._ensure_authenticated() + + assert conn._jwt_authenticated is True + assert conn._session.headers["Authorization"] == "Bearer jwt-token" + mock_post.assert_called_once() + + @patch("requests.Session.get") + @patch("requests.Session.post") + def test_get_odata_service_document(self, mock_post, mock_get): + """Test fetching and caching the OData service document.""" + post_response = MagicMock() + post_response.status_code = 200 + post_response.json.return_value = {"access_token": "jwt-token", "token_type": "bearer"} + mock_post.return_value = post_response + + get_response = MagicMock() + get_response.status_code = 200 + get_response.json.return_value = { + "value": [ + {"name": "public.examples.planets", "kind": "EntitySet", "source": "Table"}, + {"name": "public.examples.a_view", "kind": "EntitySet", "source": "View"}, + ] + } + mock_get.return_value = get_response + + conn = dbapi.Connection(host="opteryx.app", username="user", token="raw-token", ssl=True) + entities = conn.get_odata_service_document() + assert entities == [ + {"name": "public.examples.planets", "kind": "EntitySet", "source": "Table"}, + {"name": "public.examples.a_view", "kind": "EntitySet", "source": "View"}, + ] + + # Second call should be served from cache, not a second HTTP request. + conn.get_odata_service_document() + assert mock_get.call_count == 1 + + @patch("requests.Session.get") + @patch("requests.Session.post") + def test_get_odata_service_document_failure_returns_empty(self, mock_post, mock_get): + """Test that a failed request degrades to an empty list rather than raising.""" + post_response = MagicMock() + post_response.status_code = 200 + post_response.json.return_value = {"access_token": "jwt-token", "token_type": "bearer"} + mock_post.return_value = post_response + mock_get.side_effect = requests.exceptions.ConnectionError("boom") + + conn = dbapi.Connection(host="opteryx.app", username="user", token="raw-token", ssl=True) + assert conn.get_odata_service_document() == [] + + @patch("requests.Session.get") + @patch("requests.Session.post") + def test_get_odata_metadata(self, mock_post, mock_get): + """Test fetching and parsing the OData $metadata document.""" + post_response = MagicMock() + post_response.status_code = 200 + post_response.json.return_value = {"access_token": "jwt-token", "token_type": "bearer"} + mock_post.return_value = post_response + + metadata_xml = """ + + + + + + + + + + + +""" + get_response = MagicMock() + get_response.status_code = 200 + get_response.content = metadata_xml.encode("utf-8") + mock_get.return_value = get_response + + conn = dbapi.Connection(host="opteryx.app", username="user", token="raw-token", ssl=True) + metadata = conn.get_odata_metadata() + assert metadata == { + "public_examples_planets": [ + ("id", "Edm.Int64", False), + ("name", "Edm.String", True), + ("mass", "Edm.Double", True), + ] + } + + # Second call should be served from cache, not a second HTTP request. + conn.get_odata_metadata() + assert mock_get.call_count == 1 + + @patch("requests.Session.get") + @patch("requests.Session.post") + def test_get_odata_metadata_malformed_xml_returns_empty(self, mock_post, mock_get): + """Test that malformed XML degrades to an empty dict rather than raising.""" + post_response = MagicMock() + post_response.status_code = 200 + post_response.json.return_value = {"access_token": "jwt-token", "token_type": "bearer"} + mock_post.return_value = post_response + + get_response = MagicMock() + get_response.status_code = 200 + get_response.content = b" SQLAlchemy type mapping, including an unknown fallback.""" + from sqlalchemy import types as sqltypes + from sqlalchemy_dialect.dialect import _edm_type_to_sqlalchemy + + assert _edm_type_to_sqlalchemy("Edm.Int64") is sqltypes.BigInteger + assert _edm_type_to_sqlalchemy("Edm.String") is sqltypes.String + assert _edm_type_to_sqlalchemy("Edm.Double") is sqltypes.Float + assert _edm_type_to_sqlalchemy("Edm.Boolean") is sqltypes.Boolean + # Unknown/unmapped Edm types fall back to String rather than raising. + assert _edm_type_to_sqlalchemy("Edm.SomeFutureType") is sqltypes.String + + def test_has_table_true(self): + """Test has_table for an entity present in the service document.""" + sa_connection, dialect_module = self._fake_sa_connection(self.SERVICE_DOCUMENT) + dialect = OpteryxDialect() + assert dialect.has_table(sa_connection, "planets", schema="public.examples") is True + + def test_has_table_false(self): + """Test has_table for an entity absent from the service document.""" + sa_connection, _ = self._fake_sa_connection(self.SERVICE_DOCUMENT) + dialect = OpteryxDialect() + assert dialect.has_table(sa_connection, "does_not_exist", schema="public.examples") is False + + def test_has_table_no_schema(self): + """Test has_table using a fully-dotted table_name with no separate schema.""" + sa_connection, _ = self._fake_sa_connection(self.SERVICE_DOCUMENT) + dialect = OpteryxDialect() + assert dialect.has_table(sa_connection, "public.examples.planets") is True + + def test_get_table_names_scoped(self): + """Test get_table_names filters by schema and strips the dotted prefix.""" + sa_connection, _ = self._fake_sa_connection(self.SERVICE_DOCUMENT) + dialect = OpteryxDialect() + tables = dialect.get_table_names(sa_connection, schema="public.examples") + assert tables == ["planets", "moons"] + + def test_get_table_names_unscoped(self): + """Test get_table_names with no schema returns full dotted names, tables only.""" + sa_connection, _ = self._fake_sa_connection(self.SERVICE_DOCUMENT) + dialect = OpteryxDialect() + tables = dialect.get_table_names(sa_connection) + assert tables == ["public.examples.planets", "public.examples.moons", "personal.bastian.customers"] + + def test_get_view_names(self): + """Test get_view_names only returns entities with source == 'View'.""" + sa_connection, _ = self._fake_sa_connection(self.SERVICE_DOCUMENT) + dialect = OpteryxDialect() + views = dialect.get_view_names(sa_connection, schema="public.examples") + assert views == ["a_view"] + + def test_get_schema_names(self): + """Test get_schema_names derives distinct dotted prefixes.""" + sa_connection, _ = self._fake_sa_connection(self.SERVICE_DOCUMENT) + dialect = OpteryxDialect() + schemas = dialect.get_schema_names(sa_connection) + assert schemas == ["personal.bastian", "public.examples"] + + def test_get_columns(self): + """Test get_columns parses $metadata into SQLAlchemy column dicts.""" + from sqlalchemy import types as sqltypes + + sa_connection, _ = self._fake_sa_connection(metadata=self.METADATA) + dialect = OpteryxDialect() + columns = dialect.get_columns(sa_connection, "planets", schema="public.examples") + + assert [c["name"] for c in columns] == ["id", "name", "mass"] + assert columns[0]["type"] is sqltypes.BigInteger + assert columns[0]["nullable"] is False + assert columns[1]["type"] is sqltypes.String + assert columns[1]["nullable"] is True + assert columns[2]["type"] is sqltypes.Float + + def test_get_columns_unknown_table(self): + """Test get_columns returns an empty list for a table absent from $metadata.""" + sa_connection, _ = self._fake_sa_connection(metadata=self.METADATA) + dialect = OpteryxDialect() + assert dialect.get_columns(sa_connection, "does_not_exist", schema="public.examples") == [] + + class TestConnect: """Tests for the connect function."""