🚀 Snowforge - Enterprise Data Integration Platform

Snowforge is a comprehensive Python package designed to streamline data integration and orchestration across cloud and on-premise systems. It provides a unified interface for managing data pipelines between AWS, Snowflake, Microsoft Fabric, Power BI, and various on-premise database systems.

The goal is straightforward: integrations should be easy to compose, explicit about configuration and failures, and useful from both small scripts and larger data-movement workflows.

The package currently requires Python 3.12 or newer and uses uv for dependency management, locked environments, and project commands.

Contents

• Capabilities
• Where Snowforge is heading
• Installation
• Configuration
• AWS
• Snowflake
• Netezza DataMover
• Snowflake execution logging
• Fabric and Power BI
• Report broadcasting
• Development
• Releasing
• Contributing
• Contact

Capabilities

Where Snowforge is heading

The extension interfaces are already in place for the next integrations:

• Teams will join email as a broadcaster delivery channel.
• JsonFileConfigProvider will support file-driven broadcast jobs.
• The QlikSenseExporter scaffold is ready for a concrete Qlik integration.
• ExtractorStrategy provides the path for Oracle, PostgreSQL, and other database extractors alongside the current Netezza implementation.

These are visible extension points rather than hidden promises: their modules and interfaces are present, while unfinished operations still fail clearly with NotImplementedError.

Installation

Add Snowforge to a uv project

uv add snowforge-package

This adds Snowforge to the project's pyproject.toml and resolves it into the project's uv.lock.

Work on Snowforge from source

git clone git@github.com:Norsk-Tipping/snowforge-package.git
cd snowforge-package
uv sync

Run commands inside the locked environment with uv run:

uv run pytest
uv run ruff check .
uv run pyrefly check

Runtime and development dependencies are declared in pyproject.toml; exact resolved versions are tracked in uv.lock.

Configuration

Snowforge looks for snowforge_config.toml in this order:

1. SNOWFORGE_CONFIG_PATH
2. ./snowforge_config.toml
3. ~/.config/snowforge_config.toml
4. The installed Snowforge package directory

Paths supplied directly to Config may be strings or pathlib.Path objects. Config.find_config_file() returns a Path or None.

[AWS.default]
AWS_ACCESS_KEY = "your-access-key-id"
AWS_SECRET_KEY = "your-secret-access-key"
REGION = "eu-west-1"

[SNOWFLAKE.default]
USERNAME = "your-username"
ACCOUNT = "your-account-identifier"
ROLE = "SYSADMIN"
SNOWFLAKE_WAREHOUSE = "COMPUTE_WH"

[SNOWFLAKE.svc_key_based]
USERNAME = "service-account"
ACCOUNT = "your-account-identifier"
KEY_FILE_PATH = "/path/to/private-key.p8"
KEY_FILE_PASSWORD = "key-password"
SNOWFLAKE_DATABASE = "ANALYTICS"
SNOWFLAKE_SCHEMA = "PUBLIC"
SNOWFLAKE_WAREHOUSE = "ETL_WH"
ROLE = "ETL_ROLE"

USERNAME and ACCOUNT are required for Snowflake profiles. KEY_FILE_PASSWORD is valid only together with KEY_FILE_PATH. All three AWS fields are required.

Use a profile named snowforge when calling SnowflakeLogging; that module connects with profile="snowforge".

AWS

Initialize the AWS clients before using Secrets Manager or S3:

from Snowforge import AWSIntegration

AWSIntegration.initialize(profile="default", verbose=True)

token = AWSIntegration.get_secret("powerbi-refresh-token")

AWSIntegration.push_file_to_s3(
    bucket_name="data-lake",
    file_to_upload="/tmp/orders.csv",
    key="raw/orders.csv",
)

for key in AWSIntegration.get_bucket_contents("data-lake") or []:
    print(key)

get_secret() returns a non-empty string. It raises when the client has not been initialized, AWS rejects the request, or the secret does not contain a non-empty SecretString.

Snowflake

Connect using a configured profile:

from Snowforge import SnowflakeIntegration

connection = SnowflakeIntegration.connect(
    profile="svc_key_based",
    verbose=True,
)

with connection.cursor() as cursor:
    cursor.execute("SELECT CURRENT_VERSION()")
    print(cursor.fetchone())

SnowflakeIntegration.close_connection()

Load CSV or Parquet data from a Snowflake stage:

SnowflakeIntegration.load_to_snowflake(
    stage="S3_STAGE",
    stage_key="/raw/orders.parquet",
    database="ANALYTICS",
    schema="RAW",
    table="ORDERS",
    profile="svc_key_based",
    file_format="parquet",
    verbose=True,
)

file_format accepts "csv" (the default) or "parquet". The Parquet path uses Snowflake's vectorized scanner and case-insensitive column matching.

SnowflakeIntegration reuses one open class-level connection for the current profile. It does not implement a connection pool.

Netezza DataMover

The Netezza extractor requires the appropriate NZ_HOME, NZ_USER, NZ_PASSWORD, and NZ_DATABASE environment variables and the nzsql executable.

from Snowforge.DataMover.DataMover import Engine
from Snowforge.DataMover.Extractors.NetezzaExtractor import NetezzaExtractor

extractor = NetezzaExtractor()

header, csv_file = Engine.export_to_file(
    extractor=extractor,
    output_path="/tmp/exports",
    fully_qualified_table_name="PROD.SALES.ORDERS",
    filter_statement="order_date >= CURRENT_DATE - 7",
    verbose=True,
)

filter_statement is a predicate only; do not include the WHERE keyword. Table names may be DATABASE.SCHEMA.TABLE, or SCHEMA.TABLE when NZ_DATABASE supplies the default database.

Engine.determine_file_offsets() avoids splitting quoted multiline records. Engine.parallel_process() starts worker processes and returns them to the caller; callers remain responsible for joining those processes.

Snowflake execution logging

The SQL required by SnowflakeLogging is packaged under Snowforge/resources/sql. Inspect it before applying it to a Snowflake account:

from Snowforge import SnowflakeLogging

sql_by_filename = SnowflakeLogging.show_requirements(print_to_console=True)

After installing the SQL objects and configuring the snowforge profile:

from datetime import datetime

from Snowforge import SnowflakeLogging

execution_id = SnowflakeLogging.log_start(
    task_id=42,
    process_id=12345,
    starttime=datetime.now(),
)

SnowflakeLogging.log_end(
    execution_id=execution_id,
    status="SUCCESS",
    log_path="/logs/task-42.log",
    endtime=datetime.now(),
    next_execution_time=None,
)

Fabric and Power BI

FabricIntegration obtains its bearer token from AWS Secrets Manager. It supports direct pipeline, semantic-model, and paginated-report operations:

from Snowforge import FabricIntegration

fabric = FabricIntegration(
    aws_secret_name="powerbi-refresh-token",
    aws_profile="default",
)

if fabric.test_authentication():
    execution_id = fabric.start_pipeline(
        workspace_id="workspace-guid",
        item_id="pipeline-guid",
    )
    print(execution_id)

fabric.semantic_model_reload(
    workspace_id="workspace-guid",
    item_id="semantic-model-guid",
)

For parameterized paginated exports, include fileformat in input_data; the Jinja template reads the format from that mapping:

response = fabric.export_paginated_report(
    access_token=fabric.access_token,
    workspace_id="workspace-guid",
    report_id="report-guid",
    input_data={
        "fileformat": "XLSX",
        "StartDate": "2026-01-01",
        "Department": "Sales",
    },
)

These methods expose status strings or raw requests.Response objects, making it possible for each workflow to apply its own retry and failure policy.

Report broadcasting

Snowforge's broadcaster separates report export, configuration, and delivery so that new sources and channels can be added without rewriting the workflow. Today it includes:

• ReportExporter and ConfigProvider extension interfaces
• FabricExporter integration adapter
• SnowflakeConfigProvider for DSB_RECIPIENT
• BroadcastConfig normalization and dynamic parameter tokens
• SMTP email delivery with optional attachments

The Fabric adapter normalizes export statuses to the broadcaster contract, so Fabric paginated-report exports can run through the complete broadcast flow.

Load configurations from Snowflake with a cursor and a job name:

from Snowforge import SnowflakeIntegration
from Snowforge.Broadcaster import SnowflakeConfigProvider

connection = SnowflakeIntegration.connect(profile="default")
provider = SnowflakeConfigProvider(
    cursor=connection.cursor(),
    table="DEV_ADMIN.DOLPHIN_BRODCASTER.DSB_RECIPIENT",
)
configs = provider.fetch("SWF_TEST_PAGINATED")

The Snowflake table supplies sender, recipients, channel, parameters, and body. It does not supply workspace_id or report_id; callers must attach those before exporting a report.

Email is the first complete delivery channel. Teams, JSON-file configuration, and Qlik export remain on the roadmap described above.

Package layout

Snowforge/
├── AWSIntegration.py
├── Config.py
├── ConfigModels.py
├── Logging.py
├── SnowflakeIntegration.py
├── SnowflakeLogging.py
├── Broadcaster/
├── DataMover/
├── FabricIntegration/
└── resources/sql/

Development

Before submitting a change, run:

uv run ruff format --check .
uv run ruff check .
uv run pyrefly check
uv run pytest
uv lock --check

The version, dependencies, package discovery, and bundled resource files are configured in pyproject.toml. setup.py is only a compatibility shim for older tooling.

Releasing

Snowforge consumers, including SnowforgeDataMover, install official releases from PyPI. Merging a change to main makes it available to source-based compatibility CI, but it does not update normal installations or their lockfiles.

To publish a release:

1. Update the project version with uv version X.Y.Z, include that change in the release PR, and merge it to main after CI passes.
2. Create and push an annotated vX.Y.Z tag on the resulting main commit.
3. Create a draft GitHub Release for that tag and review its notes.
4. Publish the GitHub Release. This triggers pypi_release.yaml, which builds the tagged source and publishes it to PyPI.

uv-managed consumers continue using the exact Snowforge version recorded in their lockfile until they deliberately update it. For a compatible release, update only the locked Snowforge package:

uv lock --upgrade-package snowforge-package

When consumer code requires a new minimum Snowforge version, let uv update both the dependency declaration and lockfile instead of editing pyproject.toml manually:

uv add "snowforge-package>=X.Y.Z"

Commit the resulting project metadata and lockfile changes together.

License

Snowforge is intended to be distributed under the MIT License, which is already declared in the package metadata. The repository's LICENSE placeholder will be populated with the applicable license text before release.

Contributing

Contributions, suggestions, and collaboration are welcome.

1. Fork the repository.
2. Create a feature branch: git checkout -b feature/my-new-feature.
3. Make the change and run the checks described in Development.
4. Commit the change: git commit -am 'Add new feature'.
5. Push the branch and submit a pull request.

For bugs and feature requests, open an issue in the GitHub repository.

Contact

For questions about Snowforge, proposals for improvement, or collaboration:

• Authors: Andreas Heggelund and Christophe Lebegue
• Email: andreasheggelund@gmail.com

Vi oppfordrer til å ta kontakt dersom du har forslag til forbedringer, spørsmål om bruken av Snowforge, eller ønsker samarbeid. Ditt bidrag er alltid velkommen!

🙏 Acknowledgments

Snowforge is maintained by Norsk Tipping and built on top of excellent open-source libraries:

• boto3 - AWS SDK for Python
• snowflake-connector-python - Snowflake connector
• Jinja2 - Template engine
• requests - HTTP library

Happy Data Engineering! 🚀