Skip to main content
Pre-release

This release is a pre-release and may not be stable for production use.

dataforge-sdk

SDK for creating DataForge extensions.

Example projects and usage patterns: https://github.com/dataforgelabs/dataforge-sdk

Postgres Utilities

The dataforge.pg module provides helper functions to execute SQL operations against the DataForge Postgres metastore:

from dataforge.pg import select, update, pull

# Execute a SELECT query and return a Spark DataFrame
df = select("SELECT * FROM my_table")

# Execute an UPDATE/INSERT/DELETE query
update("UPDATE my_table SET col = 'value'")

# Trigger a new data pull for source_id 123
pull(123)

IngestionSession

The IngestionSession class manages a custom data ingestion process lifecycle.

from dataforge import IngestionSession

# Initialize a session (production use)
session = IngestionSession()

# Initialize a session (optional source_name/project_name for testing)
session = IngestionSession(source_name="my_source", project_name="my_project")

# Ingest data 
# pass a function returning a DataFrame (recommended to integrate logging with DataForge)
session.ingest(lambda: spark.read.csv("s3://bucket/path/input.csv"))

# pass a DataFrame (can be used for testing, not recommended for production deployment)
df = spark.read.csv("s3://bucket/path/input.csv")
session.ingest(df)

# ingest empty dataframe to create 0-record input
session.ingest()


# Fail the process with error message
session.fail("Error message")

# Retrieve latest tracking fields
tracking = session.latest_tracking_fields()

# Retrieve connection parameters for the current source
connection_parameters = session.connection_parameters()

# Retrieve custom parameters for the current source
custom_parameters = session.custom_parameters()

# Retrieve system configuration for the current session
system_configuration = session.system_configuration

ParsingSession

The ParsingSession class manages a custom parse process lifecycle.

from dataforge import ParsingSession

# Initialize a session (production use)
session = ParsingSession()

# Initialize a session (optional input_id for testing)
session = ParsingSession(input_id=123)

# Retrieve custom parameters
params = session.custom_parameters()

# Retrieve system configuration for the current session
system_configuration = session.system_configuration

# Get the path of file to be parsed
path = session.file_path

# Run parsing: pass a DataFrame, a function returning a DataFrame or None (0-record file)
session.run(lambda: spark.read.json(session.file_path))

# Fail the process with error message
session.fail("Error message")

PostOutputSession

The PostOutputSession class manages a custom post-output process lifecycle.

from dataforge import PostOutputSession

# Initialize a session (production use)
session = PostOutputSession()

# Initialize a session (optional names for testing)
session = PostOutputSession(output_name="report", output_source_name="my_source", project_name="my_project")


# Get the path of file generated by preceding output process
path = session.file_path()

# Retrieve connection parameters for the current output
connection_parameters = session.connection_parameters()

# Retrieve custom parameters for the current source
custom_parameters = session.custom_parameters()

# Retrieve system configuration for the current session
system_configuration = session.system_configuration

# Run post-output logic: pass a function encapsulating custom code
session.run(lambda: print(f"Uploading file from {path}"))

# Fail the process with error message
session.fail("Error message")

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

dataforge_sdk-10.3.0rc16.tar.gz (13.8 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

dataforge_sdk-10.3.0rc16-py3-none-any.whl (20.4 kB view details)

Uploaded Python 3

File details

Details for the file dataforge_sdk-10.3.0rc16.tar.gz.

File metadata

  • Download URL: dataforge_sdk-10.3.0rc16.tar.gz
  • Upload date:
  • Size: 13.8 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/7.0.0 CPython/3.10.20

File hashes

Hashes for dataforge_sdk-10.3.0rc16.tar.gz
Algorithm Hash digest
SHA256 23a03862f6989e8cd9e508a9e3b49c6360a0721d87e6b7c1456cda99bc3e083c
MD5 563014a48f895b785d44d819fef75d88
BLAKE2b-256 8f86570a87271b3f3885714532dbe6e35b34f461e8154f997ebc19c1915c040b

See more details on using hashes here.

File details

Details for the file dataforge_sdk-10.3.0rc16-py3-none-any.whl.

File metadata

File hashes

Hashes for dataforge_sdk-10.3.0rc16-py3-none-any.whl
Algorithm Hash digest
SHA256 87a9a07590f1dfbf310281c2f76972ead336cc034febb5aa820272d8cf7c46e5
MD5 0a2c67aab19cbeb536cc62c97ac967c3
BLAKE2b-256 fb3d8f3958b11ae9c781eb4d12810e9b0f62276fb6a3a37278a7f32338b49aba

See more details on using hashes here.

Release history Release notifications | RSS feed

10.3.0

2 files

This release

10.3.0rc16 This release

2 files

10.2.0

2 files

10.1.1

2 files

10.1.0

2 files

10.0.3

2 files

10.0.2

2 files

10.0.1

2 files

10.0.0

2 files

9.2.6

2 files

9.2.5

2 files

9.2.4

2 files

9.2.3

2 files

9.2.2

2 files

9.2.1

2 files

9.2.0

2 files

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page