Feat/hana engine

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- `hana` server type for `datacontract test`, supporting SAP HANA Cloud and SAP Datasphere (Open SQL schemas). Install with `pip install datacontract-cli[hana]`. (`hdbcli` is under SAP's proprietary license and is therefore not part of `[all]`.)
+
 ## [1.0.4] - 2026-06-22
 
 ### Added

README.md

@@ -1,4 +1,4 @@
-# Data Contract CLI
+# Data Contract CLI
 
 <p>
   <a href="https://github.com/datacontract/datacontract-cli/actions/workflows/ci.yaml?query=branch%3Amain">
@@ -267,6 +267,7 @@ A list of available extras:
 | RDF                                      | `pip install datacontract-cli[rdf]`        |
 | Amazon Redshift                          | `pip install datacontract-cli[redshift]`   |
 | S3 Integration                           | `pip install datacontract-cli[s3]`         |
+| SAP HANA / Datasphere                    | `pip install datacontract-cli[hana]`       |
 | Snowflake Integration                    | `pip install datacontract-cli[snowflake]`  |
 | Microsoft SQL Server                     | `pip install datacontract-cli[sqlserver]`  |
 | Trino                                    | `pip install datacontract-cli[trino]`      |

datacontract/engines/data_contract_test.py

@@ -64,6 +64,12 @@ def execute_data_contract_test(
     if server.type == "api":
         server = process_api_response(run, server)
 
+    if server.type == "hana":
+        from datacontract.engines.hana.check_hana_execute import check_hana_execute
+
+        check_hana_execute(run, data_contract, server, schema_name=schema_name, check_categories=check_categories)
+        return
+
     specs = create_checks(data_contract, server, schema_name=schema_name)
     if check_categories is not None:
         specs = [s for s in specs if s.category in check_categories]

datacontract/engines/hana/__init__.py

@@ -0,0 +1 @@
+

datacontract/engines/hana/check_hana_execute.py

@@ -0,0 +1,59 @@
+from open_data_contract_standard.model import OpenDataContractStandard, Server
+
+from datacontract.engines.hana.hana_connection import get_connection
+from datacontract.engines.hana.hana_quality_check import run_quality_checks, run_sla_checks
+from datacontract.engines.hana.hana_schema_check import run_schema_checks
+from datacontract.model.exceptions import DataContractException
+from datacontract.model.run import ResultEnum, Run
+
+
+def check_hana_execute(
+    run: Run,
+    data_contract: OpenDataContractStandard,
+    server: Server,
+    schema_name: str = "all",
+    check_categories: set[str] | None = None,
+):
+    connection = None
+    try:
+        run.log_info("Running engine hana")
+        server_schema = server.schema_
+        if not server_schema:
+            raise DataContractException(
+                type="hana-connection",
+                name="Missing SAP HANA schema",
+                result=ResultEnum.failed,
+                reason="Server schema is required for SAP HANA Cloud.",
+                engine="hana",
+            )
+        connection = get_connection(server)
+        checks_before = len(run.checks)
+
+        for schema_object in data_contract.schema_ or []:
+            if schema_name != "all" and schema_object.name != schema_name:
+                continue
+            if check_categories is None or "schema" in check_categories:
+                run.checks.extend(run_schema_checks(connection, server_schema, schema_object))
+            if check_categories is None or "quality" in check_categories:
+                run.checks.extend(run_quality_checks(connection, server_schema, schema_object))
+
+        if check_categories is None or "servicelevel" in check_categories:
+            run.checks.extend(run_sla_checks(connection, server_schema, data_contract, schema_filter=schema_name))
+
+        if check_categories is not None and len(run.checks) == checks_before:
+            run.log_warn(f"No checks found for categories: {', '.join(sorted(check_categories))}")
+
+    except DataContractException:
+        raise
+    except Exception as e:
+        raise DataContractException(
+            type="hana",
+            name="SAP HANA Cloud test execution error",
+            result=ResultEnum.error,
+            reason=str(e),
+            engine="hana",
+            original_exception=e,
+        )
+    finally:
+        if connection is not None:
+            connection.close()

datacontract/engines/hana/hana_connection.py

@@ -0,0 +1,66 @@
+import os
+from contextlib import contextmanager
+from typing import Any
+
+from open_data_contract_standard.model import Server
+
+from datacontract.model.exceptions import DataContractException, require_env
+from datacontract.model.run import ResultEnum
+
+
+def _env_bool(name: str, default: str = "true") -> bool:
+    return os.getenv(name, default).lower() in ("true", "1", "yes")
+
+
+def import_hdbcli():
+    try:
+        from hdbcli import dbapi
+
+        return dbapi
+    except ImportError as e:
+        raise DataContractException(
+            type="hana-connection",
+            result=ResultEnum.failed,
+            name="hdbcli is missing",
+            reason="Install the extra datacontract-cli[hana] to use SAP HANA Cloud.",
+            engine="hana",
+            original_exception=e,
+        )
+
+
+def get_connection(server: Server) -> Any:
+    dbapi = import_hdbcli()
+    username = require_env("DATACONTRACT_HANA_USERNAME", server_type="hana")
+    password = require_env("DATACONTRACT_HANA_PASSWORD", server_type="hana")
+    encrypt = _env_bool("DATACONTRACT_HANA_ENCRYPT")
+    ssl_validate = _env_bool("DATACONTRACT_HANA_SSL_VALIDATE_CERTIFICATE")
+    ssl_hostname = os.getenv("DATACONTRACT_HANA_SSL_HOSTNAME_IN_CERTIFICATE", "*")
+
+    try:
+        return dbapi.connect(
+            address=server.host,
+            port=server.port or 443,
+            user=username,
+            password=password,
+            encrypt=encrypt,
+            sslValidateCertificate=ssl_validate,
+            sslHostnameInCertificate=ssl_hostname,
+        )
+    except Exception as e:
+        raise DataContractException(
+            type="hana-connection",
+            result=ResultEnum.failed,
+            name="SAP HANA Cloud connection error",
+            reason=str(e),
+            engine="hana",
+            original_exception=e,
+        )
+
+
+@contextmanager
+def hana_connection(server: Server):
+    connection = get_connection(server)
+    try:
+        yield connection
+    finally:
+        connection.close()