Platform¶

Platform abstraction: auto-detects Fabric, Databricks, or vanilla Spark and exposes a single shared interface for config reading, Delta writes, and filesystem operations regardless of which platform is actually running.

class scamander.processor.platform.Platform(*args, **kwargs)¶

Bases: ABC

Abstract base class that defines the shared interface for all supported Spark platforms.

On instantiation, the platform is auto-detected from environment variables and a concrete subclass (Fabric, Databricks, or Spark) is returned transparently via __new__. Shared logic — config reading, Delta writes, filename parsing, and datatype mapping — lives here so platform-specific subclasses stay lean.

Subclasses must implement:

validate_compatibility – assert that the environment meets platform requirements.
check_file_exists – use the platform’s native filesystem API.
resolve_filemask – expand a glob/wildcard path into concrete file paths.
_list_dir_names – list base names of entries directly inside a folder, used by resolve_filemask_recursive.
move_file – move/rename a file via the platform’s filesystem API.
get_notebook_name – return the name of the currently running notebook.
get_notebook_url – return a browser-accessible URL to the notebook.
create_empty_table_from_metadata – DDL-create a Delta table from a metadata dict.

Parameters:

logging_catalog (str) – Unity Catalog / catalog name that contains the logging schema. Defaults to "main".
logging_schema (str) – Schema name for the logging Delta table. Defaults to "mdif".
logging_table (str) – Table name used to persist machine-readable log records. Defaults to "logging".
retention_days (int) – Number of days of machine log history to retain. Older rows are deleted on every Platform initialisation. Defaults to 90.
log_level (str) – Minimum level a record must meet to be written to the machine log table (e.g. "ERROR"). Independent of what’s printed to the console. Defaults to "INFO".
log_timezone (str) – IANA timezone name used to stamp machine log timestamps. Defaults to "Europe/Amsterdam".
log_flush_interval_seconds (int) – How often the machine log’s background timer auto-flushes buffered rows. Set to 0 to disable the timer and rely only on error-triggered/explicit/atexit flushes. Defaults to 300 (5 minutes).

Example:

from scamander import Platform

# Auto-detects Fabric / Databricks / vanilla Spark from env vars.
platform = Platform()
platform.log.info("Pipeline started")

abstractmethod check_file_exists(file_name: str, raise_exceptions=False) → bool¶

Check whether a file or directory exists on the platform’s filesystem.

Parameters:

file_name – Absolute path to the file or directory.
raise_exceptions – When True, raise FileNotFoundError instead of returning False if the path is missing. Defaults to False.

Returns:

True if the path exists, False otherwise.

Return type:

bool

Raises:

FileNotFoundError – Only when raise_exceptions=True and the path does not exist.

check_file_set_exists(config: list[dict], file_key: str, raise_exceptions: bool = False) → bool¶

Check that every file path listed in a config exists on the platform filesystem.

Iterates each record in config, reads the file path from record[file_key], and calls check_file_exists() for each. All records are checked before raising so the caller gets a complete picture of what is missing.

Parameters:

config – Parsed configuration — a list of dicts, as returned by read_config() for CSV, JSON array, or YAML sequence sources.
file_key – The dict key whose value holds the file path to check.
raise_exceptions – When True, raises FileNotFoundError after checking all records if any files are missing. Defaults to False.

Returns:

True if every file exists, False if one or more are missing.

Return type:

bool

Raises:

FileNotFoundError – Only when raise_exceptions=True and at least one file is missing. The message lists all missing paths.
KeyError – If a record in config does not contain file_key.

abstractmethod create_empty_table_from_metadata(metadata: dict)¶

DDL-create an empty Delta table from a metadata definition.

Column definitions are generated via map_datatype(). When metadata["track_history"] is True, the five SCD Type-2 columns (HashSleutel, HashWaarde, GeldigVan, GeldigTot, IsActief) are appended automatically. Platforms that support Delta constraints (e.g. Databricks Unity Catalog) may also emit PRIMARY KEY / FOREIGN KEY clauses from the metadata.

Parameters:: metadata – Metadata dictionary with "table", "columns", and "track_history" keys — see map_datatype() for the shape of each entry in "columns".

static extract_datetime_from_filename(filename: str) → datetime¶

Extracts a datetime from a filename by searching for a 12-digit timestamp in the format YYYYMMDDHHMI (year, month, day, hour, minute).

This is useful for sorting a list of files resolved by resolve_filemask in chronological order so that the most-recent file can be selected or processed last.

Parameters:

filename (str) – The filename (or full path) to inspect.

Returns:

The extracted datetime if a valid 12-digit sequence is found;: datetime.max otherwise, so the file naturally sorts to the end of any ordered list.

Return type:

datetime

Example:

>>> Platform.extract_datetime_from_filename("export_202503151430_v1.csv")
datetime(2025, 3, 15, 14, 30)

flush_logs() → None¶

Force any buffered machine-log rows to be written to the Delta logging table.

MachineLogger buffers low-severity log rows in memory for performance and only flushes automatically on error, at process exit, or via the installed sys.excepthook. Call this explicitly (e.g. in a notebook’s final cell) to guarantee the full log trail is persisted before the run ends.

abstractmethod get_notebook_name() → str¶

Return the name of the currently running notebook.

Used to label human/machine log records with their origin. Returns a platform-appropriate placeholder (e.g. "local") when no notebook context is available.

Returns:: The notebook’s display name.
Return type:: str

abstractmethod get_notebook_url() → str¶

Return a browser-accessible URL to the currently running notebook.

Used to link machine log records back to their source notebook. Returns a platform-appropriate placeholder (e.g. "local") when no notebook context is available.

Returns:: The notebook’s URL.
Return type:: str

get_secret(secret_name: str, vault: str | None = None) → str¶

Retrieve a secret via the configured SecretProvider.

Delegates to the SecretProvider passed during construction. When no provider was supplied, the call raises NotImplementedError with instructions on how to configure one.

Parameters:

secret_name – Name or path of the secret.
vault – Provider-specific vault / scope / mount identifier. See the concrete SecretProvider for details.

Returns:

The secret value as a plain string.

Raises:

NotImplementedError – When no SecretProvider was configured.

static map_datatype(col: dict) → str¶

Converts a metadata column definition into a Delta Lake column DDL fragment.

Accepts a wide range of SQL Server, PostgreSQL, and generic SQL type names and normalises them to their Delta / Spark SQL equivalents. The mapping is intentionally broad so that metadata config files can use source-system type names directly.

Parameters:

col (dict) –

A column metadata dictionary with at minimum: - "target" (str): The column name in the Delta table. - "type" (str): The source data type string. - "nullable" (bool, optional): Whether the column allows NULLs.

Defaults to True.

"precision" (int, optional): Decimal precision (DECIMAL only).
"scale" (int, optional): Decimal scale (DECIMAL only).

Returns:

A SQL DDL column fragment, e.g. "amount DECIMAL(18,2) NOT NULL".

Return type:

str

abstractmethod move_file(source: str, destination: str, overwrite=True, create_missing_dirs: bool = False) → bool¶

Move/rename a file via the platform’s native filesystem API.

Parameters:

source – Absolute path to the source file.
destination – Absolute destination path.
overwrite – When True (default), replace an existing file at destination.
create_missing_dirs – When True, create the destination directory if it does not exist. Defaults to False.

Returns:

True on success, False on any failure.

Return type:

bool

read_config(config_path: str)¶

Reads a configuration file from the platform’s filesystem and returns its contents.

Supported formats:

.json – read directly via the Hadoop filesystem and parsed with json.loads; returned as a dict or list.
.csv – parsed via Spark with a header row; returned as a list of dicts (one dict per row, via pandas to_dict(orient="records")).
.yaml / .yml – read directly via the Hadoop filesystem and parsed with yaml.safe_load; returned as a dict or list.

Parameters:

config_path (str) – Absolute path (including filename and extension) to the configuration file on the platform filesystem (e.g. DBFS, OneLake).

Returns:

Parsed configuration content.

Return type:

dict | list

Raises:

FileNotFoundError – If check_file_exists reports the file is missing.
ValueError – If the file extension is not one of the supported formats.

abstractmethod resolve_filemask(mask_path: str, case_sensitive: bool = False, full_path: bool = False)¶

Expand a wildcard path into the names of matching files.

Parameters:

mask_path – Absolute path containing a * wildcard, e.g. "/data/exports/sales_*.csv".
case_sensitive – When True, match file names exactly as cased. Defaults to False (case-insensitive matching, consistent across platforms/OSes).
full_path – When True, return each match prefixed with the folder part of mask_path. Defaults to False (base names only).

Returns:

Sorted list of matching file base names, or full paths when full_path=True. Empty if nothing matches or the parent directory is missing.

Return type:

list[str]

resolve_filemask_recursive(mask_path: str, case_sensitive: bool = False) → list[str]¶

Expand a wildcard path into full paths of matching files, walking wildcarded folders too.

Unlike resolve_filemask(), which only expands a wildcard in the final path segment, this method also expands * wildcards in intermediate folder segments — e.g. "/path/to/*csv*/test_*.csv" first resolves *csv* against actual subfolder names, then matches test_*.csv inside each matching subfolder. Each /-separated segment from the first one containing a wildcard onward is resolved in turn via _list_dir_names(), narrowing the set of candidate paths at every step.

Because the same file name can legitimately exist under multiple matched subfolders, this returns full paths rather than base names.

Parameters:

mask_path – Absolute path containing one or more * wildcards, which may appear in any path segment, e.g. "/data/exports/*csv*/sales_*.csv".
case_sensitive – When True, match names exactly as cased. Defaults to False (case-insensitive matching, consistent across platforms/OSes).

Returns:

Sorted list of full matching file paths. Empty if no candidate matches at any level.

Return type:

list[str]

Raises:

ValueError – If mask_path contains no * wildcard.

Example:

>>> platform.resolve_filemask_recursive(
...     "/data/exports/*csv*/sales_*.csv"
... )
['/data/exports/2024_csv_final/sales_202403150900.csv']

rows_to_table(rows: list[dict], table_name: str, mode: str = 'overwrite', merge_keys: list[str] | None = None) → bool¶

Convert a list of flat dicts to a DataFrame and write to a Delta table.

All values are coerced to str (or None) so the resulting schema is uniformly StringType. An _ingestion_timestamp audit column is added automatically before writing.

Parameters:

rows – Flat dicts (values must be scalars or None).
table_name – Fully qualified Delta table name (schema.table).
mode – "overwrite", "append", or "upsert". When "upsert" is passed, merge_keys must also be provided.
merge_keys – Key columns for the upsert merge condition. Ignored unless mode="upsert".

Returns:

True on success, False on failure.

property secret_provider: SecretProvider¶: The active SecretProvider instance.

upsert_table(df: DataFrame, full_table_name: str, merge_keys: list) → bool¶

Upsert a DataFrame into a Delta table using a simple match-all merge.

Matched rows (by merge_keys) are updated column-by-column; unmatched rows are inserted. The source DataFrame doesn’t need to carry every column the target table has — this matters for sources like Microsoft Graph delta queries, which only return the properties that actually changed. Columns the target has but the source lacks are left untouched on update, and filled with NULL on insert. If the table does not yet exist it is created via a plain overwrite.

This is a simple upsert, not SCD Type-2. Use scamander.transformations.scd.apply_scd2_merge when history tracking is required.

Parameters:

df – Source DataFrame containing the latest state.
full_table_name – Fully qualified Delta table name (schema.table).
merge_keys – Column names used to match source rows to target rows.

Returns:

True on success, False on failure.

Raises:

ImportError – When delta-spark is not available.

abstractmethod validate_compatibility()¶

Assert that the current environment satisfies this platform’s requirements.

Called once, at the start of __init__(), before any logging or secret-provider setup happens. Concrete subclasses should raise if a required runtime dependency (e.g. dbutils, notebookutils) is missing so failures surface immediately at construction time rather than on first use.

Raises:: RuntimeError – If the environment does not meet this platform’s requirements. The exact condition is subclass-specific.

validate_dataframe(df: DataFrame, suite_config: QualitySuiteConfig) → QualityResult¶

Validate a DataFrame against a quality suite and log the result.

Delegates to QualityChecker using this platform’s SparkSession and logger.

Parameters:

df – DataFrame to validate.
suite_config – Quality suite definition.

Returns:

QualityResult with per-expectation outcomes. When suite_config.raise_on_failure is True a DataQualityError is raised instead of returning on failure.

write_to_table(df: DataFrame, full_table_name: str, mode: str = 'overwrite', partition_by: list[str] | None = None, merge_schema: bool = False) → bool¶

Writes a Spark DataFrame to a Delta table, creating the schema if it does not yet exist.

The table name must be fully qualified as schema.table (two-part name). If the schema / database does not exist it is created automatically with CREATE DATABASE IF NOT EXISTS before the write is attempted.

Parameters:

df – Spark DataFrame to write.
full_table_name – Full table name in the form schema.table.
mode – Write mode: "overwrite", "append", "error", or "ignore". Defaults to "overwrite".
partition_by – Column names to physically partition the Delta table by. Defaults to None (no partitioning).
merge_schema – When True, sets the Delta mergeSchema write option so new or changed columns in df are merged into the target table’s schema instead of failing the write. Defaults to False.

Returns:

True if successful, False otherwise.

Return type:

bool

Example:

platform.write_to_table(df, "silver.orders", mode="append")

platform.write_to_table(
    df,
    "gold.revenue_by_region",
    mode="append",
    partition_by=["year", "month"],
    merge_schema=True,
)

class scamander.processor.platform.SupportedPlatforms(*values)¶

Bases: Enum

Enumeration of the Spark platforms Platform can auto-detect.

unknown¶: No Fabric or Databricks environment variables were found; Platform falls back to the vanilla-Spark implementation.

fabric¶: Running inside a Microsoft Fabric notebook/pipeline.

databricks¶: Running inside an Azure/AWS Databricks cluster.