Usage guide
============

1. Initialise the platform
----------------------------

.. code-block:: python

   from scamander.processor.platform import Platform

   # Auto-detects Fabric, Databricks, or vanilla Spark.
   # Sets up both human-readable and machine-readable logging automatically —
   # console log calls are forwarded into the machine log table too.
   platform = Platform(
       logging_catalog="main",             # Unity Catalog / catalog name (default: "main")
       logging_schema="mdif",              # Schema for the logging table (default: "mdif")
       logging_table="logging",            # Delta table name (default: "logging")
       retention_days=90,                  # Machine log rows older than this are deleted on init (default: 90)
       log_level="INFO",                   # Minimum level written to the machine log table (default: "INFO")
       log_timezone="Europe/Amsterdam",    # Timezone for machine log timestamps (default: "Europe/Amsterdam")
   )

.. _mdif-read-config:

2. Read a config file
-----------------------

.. code-block:: python

   # JSON — returns a dict or list
   config_data = platform.read_config(
       "abfss://container@account.dfs.core.windows.net/configs/ingest.json"
   )

   # CSV — returns a list of dicts, one per row
   config_data = platform.read_config("dbfs:/mnt/configs/mappings.csv")

   # YAML — returns a dict or list
   config_data = platform.read_config(
       "abfss://container@account.dfs.core.windows.net/configs/pipeline.yaml"
   )

``read_config`` validates that the file exists before reading and raises
``FileNotFoundError`` if it is missing.

3. Read data from a source
-----------------------------

**Using the** ``Source`` **factory (recommended)**: ``Source(config)``
inspects ``config.source.type`` and returns the correct connector
automatically, so your code never needs to know which connector is in use.

.. code-block:: python

   from scamander.models import Config
   from scamander.models.source_config import SourceConfig
   from scamander.models.source_config.csv import CsvSourceOptions
   from scamander.models.sink_config import SinkConfig
   from scamander.models.sink_config.lakehouse import LakeHouseSinkOptions
   from scamander.sources import Source

   config = Config(
       source=SourceConfig(type="csv", options=CsvSourceOptions()),
       sink=SinkConfig(type="LakeHouse", options=LakeHouseSinkOptions()),
   )

   connector = Source(config)   # returns a CsvConnector
   df = connector.connect()     # returns a Spark DataFrame

**Instantiating a connector directly:**

.. code-block:: python

   from scamander.sources.csv_connector import CsvConnector

   connector = CsvConnector(
       file_path="abfss://raw@account.dfs.core.windows.net/exports/sales_20250315.csv",
       header=True,
       separator=",",
   )
   df = connector.connect()

Oracle and PostgreSQL sources connect over JDBC. For Oracle, both
``service_name`` and ``sid`` are supported (``service_name`` takes
precedence when both are set), and a ``query`` is automatically wrapped in
a subquery alias for JDBC compatibility:

.. code-block:: python

   from scamander.sources.oracle_connector import OracleConnector
   from scamander.models.source_config.oracle import OracleSourceOptions

   oracle_options = OracleSourceOptions(
       host="oracle.example.com",
       port=1521,
       service_name="ORCL",   # or sid="ORCL" for legacy SID-style connections
       username="hr",
       password="s3cret",
       table="HR.EMPLOYEES",  # or query="SELECT * FROM HR.EMPLOYEES WHERE active=1"
       fetch_size=10000,       # rows per JDBC round-trip (default)
   )

4. Write data to a Delta table
---------------------------------

.. code-block:: python

   # mode defaults to "overwrite"; schema is created automatically if missing
   success = platform.write_to_table(df, full_table_name="bronze.customer", mode="append")

   # Supported modes: "overwrite" (default), "append", "error", "ignore"
   platform.write_to_table(df, "silver.orders", mode="overwrite")

5. Create a table from metadata
-----------------------------------

.. code-block:: python

   metadata = platform.read_config("/configs/customer_table.json")

   # Generates and executes: CREATE TABLE IF NOT EXISTS ... USING DELTA
   platform.create_empty_table_from_metadata(metadata)

With SCD Type-2 history tracking enabled in the config (``"track_history":
true``), the five SCD Type-2 columns ``HashSleutel``, ``HashWaarde``,
``GeldigVan``, ``GeldigTot``, and ``IsActief`` are appended to the generated
DDL automatically — see :doc:`api-reference` for the full column
definitions.

6. Bronze → Silver transformations
--------------------------------------

Use ``Transformer`` to clean and standardise raw bronze data before writing
to silver. All steps preserve Spark's lazy evaluation: no actions trigger
until ``.df`` is consumed by a write.

.. code-block:: python

   from scamander.transformations import Transformer

   bronze_df = platform.spark.read.table("bronze.customer")
   metadata  = platform.read_config("/configs/customer_table.json")

   silver_df = (
       Transformer(bronze_df, log=platform.log)
       .cast_to_schema(metadata["columns"])                              # cast to metadata types
       .rename_columns({"klant_id": "customer_id", "klant_naam": "customer_name"})
       .trim_strings()
       .replace_empty_strings()
       .filter_nulls(["customer_id"])                                    # drop rows missing the PK
       .drop_duplicates(subset=["customer_id"])                          # dedup on the natural key
       .add_audit_columns(source_file="customer_export.csv")             # lineage columns
       .df
   )

   platform.write_to_table(silver_df, "silver.customer", mode="overwrite")

.. list-table:: Available cleaning methods
   :header-rows: 1

   * - Method
     - Description
   * - ``cast_to_schema(metadata)``
     - Cast columns to types declared in a metadata column list
   * - ``rename_columns(mapping)``
     - Rename via a ``{source: target}`` dict
   * - ``select_columns(columns)``
     - Drop all columns not in the list
   * - ``trim_strings()``
     - Strip leading/trailing whitespace from all STRING columns
   * - ``replace_empty_strings(replacement)``
     - Convert ``""`` to null (or a custom value) in all STRING columns
   * - ``filter_nulls(columns)``
     - Remove rows where any listed column is null
   * - ``drop_duplicates(subset)``
     - Deduplicate; scope to a subset of columns when provided
   * - ``standardize_column_names()``
     - Lowercase and snake_case all column names
   * - ``add_audit_columns(source_file, batch_id)``
     - Append ``_ingestion_timestamp``, ``_source_file``, ``_batch_id``

7. Silver → Gold transformations
------------------------------------

.. code-block:: python

   orders_df    = platform.spark.read.table("silver.orders")
   customers_df = platform.spark.read.table("silver.customer")

   gold_df = (
       Transformer(orders_df, log=platform.log)
       .join(customers_df, on="customer_id", how="left", broadcast_right=True)
       .add_derived_columns({"revenue_incl_vat": "revenue * 1.21"})
       .aggregate(
           group_by=["region", "product_category"],
           aggregations=[
               {"column": "revenue",  "function": "sum",           "alias": "total_revenue"},
               {"column": "order_id", "function": "count_distinct", "alias": "unique_orders"},
               {"column": "revenue",  "function": "avg",           "alias": "avg_order_value"},
           ],
       )
       .df
   )

   platform.write_to_table(gold_df, "gold.revenue_by_region", mode="overwrite")

.. list-table:: Available enrichment methods
   :header-rows: 1

   * - Method
     - Description
   * - ``join(other, on, how, broadcast_right)``
     - Join with another DataFrame; broadcast hint for small lookup tables
   * - ``aggregate(group_by, aggregations)``
     - GroupBy + multi-function aggregation from a spec list
   * - ``add_derived_columns(expressions)``
     - Add columns from Column objects or SQL expression strings
   * - ``deduplicate_with_ranking(partition_by, order_by, descending)``
     - Keep rank-1 row per partition via ``row_number()``
   * - ``pivot_table(group_by, pivot_col, agg_col, agg_func, pivot_values)``
     - Pivot a column into multiple columns

Supported aggregation functions: ``sum``, ``count``, ``avg``, ``mean``,
``min``, ``max``, ``first``, ``last``, ``count_distinct``, ``stddev``,
``variance``, ``collect_list``, ``collect_set``.

8. SCD Type-2 merges
------------------------

Use ``apply_scd2_merge`` when the target table has ``track_history: true``
and you want to version entity changes instead of overwriting:

.. code-block:: python

   from scamander.transformations.scd import apply_scd2_merge

   # source_df contains the latest state of each entity
   counts = apply_scd2_merge(
       spark=platform.spark,
       source_df=silver_df,
       target_table="silver.customer",
       key_columns=["customer_id"],                              # uniquely identifies an entity
       tracked_columns=["customer_name", "region", "revenue"],   # changes here create new versions
   )

   print(f"Expired: {counts['closed']}  Inserted: {counts['inserted']}")

The merge runs in two steps:

1. **Expire** active rows whose tracked-column hash has changed
   (``IsActief = false``, ``GeldigTot = now``).
2. **Insert** new rows for all source records that have no currently active
   version (covers new entities and just-expired records).

``compute_hash`` is also available standalone when you need SHA-256 hashes
in other contexts:

.. code-block:: python

   from scamander.transformations.scd import compute_hash

   df = compute_hash(df, columns=["customer_id"], alias="HashSleutel")
   df = compute_hash(df, columns=["name", "region"], alias="HashWaarde")

9. Config-driven transformation pipelines
----------------------------------------------

Every transformation step can be described in a JSON/YAML config file and
parsed via ``TransformationConfig``:

.. code-block:: json

   {
     "source_table": "bronze.customer",
     "target_table": "silver.customer",
     "column_mappings": [
       { "source": "klant_id",   "target": "customer_id",   "type": "INT",     "nullable": false },
       { "source": "klant_naam", "target": "customer_name", "type": "VARCHAR"                    }
     ],
     "trim_strings": true,
     "replace_empty_strings": true,
     "filter_null_columns": ["customer_id"],
     "drop_duplicates": true,
     "duplicate_subset": ["customer_id"],
     "add_audit_columns": true,
     "write_mode": "overwrite"
   }

.. code-block:: python

   from scamander.models import TransformationConfig

   raw = platform.read_config("/configs/bronze_to_silver_customer.json")
   cfg = TransformationConfig(**raw)

For silver → gold with joins and aggregations, the config additionally
supports ``joins``, ``group_by``, and ``aggregations`` keys.

10. File utilities
----------------------

.. code-block:: python

   # Check if a file exists — Returns True / False, or raises immediately with raise_exceptions=True
   exists = platform.check_file_exists("abfss://container@account.dfs.core.windows.net/data/file.csv")
   platform.check_file_exists("dbfs:/mnt/raw/file.csv", raise_exceptions=True)

   # Resolve a wildcard file mask — returns a sorted list of matching filenames
   files = platform.resolve_filemask(
       "abfss://raw@account.dfs.core.windows.net/exports/sales_*.csv"
   )

   # Pick the most-recent file using an embedded timestamp
   # (filenames must contain a 12-digit YYYYMMDDHHMI timestamp)
   files.sort(key=Platform.extract_datetime_from_filename)
   latest_file = files[-1]

   # Move a file
   platform.move_file(
       source="abfss://raw@account.dfs.core.windows.net/inbox/data.csv",
       destination="abfss://raw@account.dfs.core.windows.net/archive/data.csv",
       overwrite=True,
       create_missing_dirs=True,
   )

**Extract an archive**: supported extensions are ``.zip``, ``.tar``,
``.tar.gz``/``.tgz``, and ``.tar.bz2``/``.tbz2``. Every archive member path
is resolved before extraction; members that would write outside the
destination directory are rejected with a ``ValueError`` (zip-slip
protection).

.. code-block:: python

   from scamander.utils import extract_archive

   extracted_files = extract_archive(
       archive_path="/lakehouse/default/Files/Inbox/export.tar.gz",
       destination="/lakehouse/default/Files/Inbox/export/",
       log=platform.log,   # optional — logs file count on completion
   )

On Fabric, combine with ``Fabric.to_local_path`` to go from an ABFS path to
a local path first:

.. code-block:: python

   from scamander.processor.fabric import Fabric
   from scamander.utils import extract_archive, read_json_records

   local_archive = Fabric.to_local_path(f"{ABFS_PREFIX}/Files/Inbox/brp_bulk.zip")
   local_dir     = local_archive.replace(".zip", "/")

   files   = extract_archive(local_archive, local_dir, log=platform.log)
   records = read_json_records(local_dir, log=platform.log)

11. Logging
---------------

The framework sets up both loggers automatically when ``Platform()`` is
initialised, and attaches a ``MachineLogHandler`` to ``Platform.log`` so
every console log call is also persisted to the machine log table (subject
to ``log_level`` filtering). Either logger can also be used independently.

**HumanLogger** — structured stdout output, suitable for notebook cells:

.. code-block:: python

   from scamander.logging.data_pipeline_logger import HumanLogger

   log = HumanLogger(name="my_pipeline")
   log.info("Starting ingestion run")
   log.warning("Column 'age' contains nulls")
   log.error("Schema mismatch detected")

**MachineLogger** — persists structured records to a Delta table for
operational observability:

.. code-block:: python

   from scamander.logging.machine_logger import MachineLogger

   mlog = MachineLogger(
       notebook_name="Ingest_Customer",
       notebook_path="https://app.fabric.microsoft.com/...",
       session_id="sess-abc-123",
       logging_catalog="main",
       logging_schema="mdif",
       logging_table="logging",
       log_level="INFO",                 # minimum level written to the table (default: "INFO")
       timezone="Europe/Amsterdam",      # timezone for the timestamp column (default: "Europe/Amsterdam")
   )

   mlog.info(message="Loaded 50,000 rows", object_name="bronze.customer", total_records=50000, status="Success")
   mlog.error(message="File not found", metadata_config_file="/configs/ingest.json", status="Failed")

Records below ``log_level`` are dropped before any Spark work happens.
This applies both to direct ``MachineLogger`` calls and to records
forwarded from ``HumanLogger``, independently of ``HumanLogger``'s own
stdlib logging level, so ``Platform(log_level="ERROR")`` still *prints*
INFO messages; they just won't reach the table.

``MachineLogger.cleanup_old_records(retention_days)`` deletes table rows
older than ``retention_days``. ``Platform()`` calls this once per
initialisation using its own ``retention_days`` argument (default ``90``);
failures are logged as a warning rather than aborting startup.

12. OAuth 2.0 API calls
----------------------------

``OAuth2Session`` is a drop-in replacement for ``requests.Session`` when the
target API requires a client-credentials token. Tokens are fetched on
construction and refreshed automatically:

.. code-block:: python

   from scamander.api.oauth2_session import OAuth2Session

   session = OAuth2Session(
       token_url="https://login.microsoftonline.com/{tenant}/oauth2/v2.0/token",
       client_id="my-app-client-id",
       client_secret="my-app-client-secret",
       token_refresh_margin=60,  # proactively refresh 60 seconds before expiry
   )

   # Use exactly like requests.Session — the Bearer token is injected automatically
   response = session.get("https://api.example.com/v1/entities")
   response = session.post("https://api.example.com/v1/entities", json={"key": "value"})

13. Data quality checks
----------------------------

Scamander integrates `Great Expectations <https://docs.greatexpectations.io>`_
out of the box, via an ephemeral (in-memory) GE context. No filesystem
store or GE project setup is required, which keeps it suitable for Fabric
and Databricks notebooks.

Quality suites are defined as ``QualitySuiteConfig`` objects (or loaded
from JSON/YAML):

.. code-block:: python

   from scamander.models.quality_config import ExpectationSpec, QualitySuiteConfig

   suite = QualitySuiteConfig(
       name="customer_bronze_quality",
       raise_on_failure=False,   # True -> raises DataQualityError on failure
       expectations=[
           ExpectationSpec(type="not_null", column="customer_id"),
           ExpectationSpec(type="unique",   column="customer_id"),
           ExpectationSpec(type="in_set",   column="status", value_set=["active", "inactive"]),
           ExpectationSpec(type="regex",    column="email",  regex=r"^[^@]+@[^@]+\.[^@]+$"),
           ExpectationSpec(type="between",  column="age",    min_value=0, max_value=120),
           ExpectationSpec(type="row_count_between", min_value=1),
           # Allow up to 2% nulls in a non-critical column
           ExpectationSpec(type="not_null", column="phone", mostly=0.98),
       ],
   )

Run it via ``platform.validate_dataframe(df, suite)``, as a terminal step in
a ``Transformer`` chain (``.validate(suite)`` — returns a ``QualityResult``,
not ``self``), or directly via ``QualityChecker(spark, log).check(df,
suite)``.

.. list-table:: Supported expectation types
   :header-rows: 1

   * - ``type``
     - Required fields
     - Description
   * - ``not_null``
     - ``column``
     - Column must contain no nulls (honoured per row via ``mostly``)
   * - ``unique``
     - ``column``
     - All values in the column must be distinct
   * - ``multi_column_unique``
     - ``columns``
     - Combination of columns must be unique per row
   * - ``between``
     - ``column``, ``min_value``, ``max_value``
     - Values must fall within the given bounds
   * - ``in_set`` / ``not_in_set``
     - ``column``, ``value_set``
     - Values must (not) be members of the given set
   * - ``regex``
     - ``column``, ``regex``
     - Values must match the regular expression
   * - ``row_count_between``
     - ``min_value`` and/or ``max_value``
     - Table row count must be within bounds
   * - ``column_exists``
     - ``column``
     - Column must be present in the DataFrame schema
   * - ``column_type``
     - ``column``, ``type_``
     - Column must have the specified Spark type (e.g. ``"StringType"``)
   * - ``foreign_key``
     - ``column``, ``ref_table``, ``ref_column``
     - Every non-null value in ``column`` must exist in ``ref_column`` of
       ``ref_table`` (a registered Spark table or temp view)
   * - ``column_comparison``
     - ``expression``
     - A SQL boolean expression must hold for every row (e.g.
       ``"start_date <= end_date"``)
   * - ``conditional_not_null``
     - ``when_column``, ``then_column``
     - When ``when_column`` is not null, ``then_column`` must also be not
       null
   * - ``custom_sql``
     - ``sql``
     - Spark-SQL query that must return zero rows to pass; use ``{batch}``
       to reference the DataFrame being validated

All column-level expectations accept an optional ``mostly`` field (float
``0.0``–``1.0``, default ``1.0``) allowing a configurable fraction of rows
to fail before the expectation is marked failed. The four SQL-backed types
(``foreign_key``, ``column_comparison``, ``conditional_not_null``,
``custom_sql``) are all-or-nothing and don't support ``mostly``.

**Tracking history with** ``QualityReporter``: persists quality run
results to Delta tables so you can track compliance over time and drill
into the exact rows that violated each check.

.. code-block:: python

   from scamander.quality import QualityChecker, QualityReporter

   checker  = QualityChecker(spark=platform.spark, log=platform.log)
   reporter = QualityReporter(
       spark=platform.spark,
       history_table="dq.quality_history",       # required — one row per expectation per run
       violations_table="dq.quality_violations",  # optional — one row per violating row
       log=platform.log,
   )

   result = checker.check(df, suite)
   run_id = reporter.save(df, result, suite, dataset_name="customers_bronze")

Violations are written for every expectation with at least one violating
row, regardless of whether it passed via ``mostly``. Schema-level checks
(``row_count_between``, ``column_exists``, ``column_type``) never produce
violation rows. See :doc:`api-reference` for the history/violations table
schemas.

14. Anonymization / pseudonymization
------------------------------------------

Scamander provides a declarative anonymization layer for protecting PII
before writing data to downstream tables. ``hash``, ``mask``, ``redact``,
and ``generalize`` use only PySpark built-ins; ``encrypt`` and ``fake``
require ``pip install "Scamander-MDIF[anonymization]"``.

.. list-table::
   :header-rows: 1

   * - Method
     - Technique
     - Reversible
   * - ``hash``
     - SHA-256 or HMAC-SHA256
     - No
   * - ``mask``
     - Character replacement with optional prefix/suffix
     - No
   * - ``redact``
     - Replace with ``null``
     - No
   * - ``generalize``
     - Bucket numeric values into labelled ranges
     - No
   * - ``encrypt``
     - AES-256-GCM authenticated encryption
     - Yes (with key)
   * - ``fake``
     - Faker-generated synthetic data
     - No

.. code-block:: python

   from scamander.models.anonymization_config import AnonymizationConfig, AnonymizationRule

   config = AnonymizationConfig(
       name="customer_pii",
       rules=[
           AnonymizationRule(column="email", method="hash"),                                      # SHA-256
           AnonymizationRule(column="bsn", method="hash", hash_key="my-secret"),                    # keyed HMAC
           AnonymizationRule(column="card_number", method="mask", visible_suffix=4),                # keep last 4
           AnonymizationRule(column="phone", method="mask", visible_prefix=3, mask_char="X"),
           AnonymizationRule(column="birth_date", method="redact"),
           AnonymizationRule(
               column="age", method="generalize",
               bins=[0.0, 18.0, 65.0, 120.0], labels=["child", "adult", "senior"],
           ),
           AnonymizationRule(column="iban", method="encrypt", encrypt_key="<base64-32-byte-key>"),  # requires [anonymization]
           AnonymizationRule(column="full_name", method="fake", faker_provider="name"),             # requires [anonymization]
       ],
   )

Generate a one-off AES-256 key with ``base64.b64encode(os.urandom(32))``.

Use ``Anonymizer(log=platform.log).apply(df, config)`` directly, or the
``.anonymize(config)`` step inside a ``Transformer`` chain (returns
``self``, so further steps can follow):

.. code-block:: python

   silver_df = (
       Transformer(bronze_df, log=platform.log)
       .cast_to_schema(metadata["columns"])
       .trim_strings()
       .filter_nulls(["customer_id"])
       .drop_duplicates(subset=["customer_id"])
       .anonymize(config)          # apply all PII rules in one step
       .add_audit_columns(source_file="customers.csv")
       .df
   )

15. Secret providers
-------------------------

Pass a ``SecretProvider`` subclass to ``Platform()`` to enable
``platform.get_secret()``. Without one, ``get_secret`` raises
``NotImplementedError``.

.. code-block:: python

   from scamander.secrets import AzureKeyVaultProvider

   platform = Platform(
       secret_provider=AzureKeyVaultProvider(default_vault="my-keyvault"),
       ...
   )
   password = platform.get_secret("pg-password")

   # Supply a custom credential (e.g. from Databricks Service Credentials):
   from databricks.sdk.credentials_provider import databricks_sdk_credentials
   credential = databricks_sdk_credentials(...)
   platform = Platform(
       secret_provider=AzureKeyVaultProvider(default_vault="my-keyvault", credential=credential),
       ...
   )

Install the required extra first, e.g. ``pip install
'Scamander-MDIF[azure-keyvault]'``. See :doc:`api-reference` for the full
list of providers (HashiCorp Vault, Azure Key Vault, Databricks secret
scopes) and their constructor options.