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.0rc25.tar.gz
(13.8 kB
view details)
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file dataforge_sdk-10.3.0rc25.tar.gz.
File metadata
- Download URL: dataforge_sdk-10.3.0rc25.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
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
5b9bad02b40ec49e33f8fa763933b9042866477d2aa26dee96d6f812712d565b
|
|
| MD5 |
ae806024a7472d94e45efa70012e21a1
|
|
| BLAKE2b-256 |
a93f7775bafdb09a2749d1a284ddb96695b37401f2641d566682d1e32dd3ccaa
|
File details
Details for the file dataforge_sdk-10.3.0rc25-py3-none-any.whl.
File metadata
- Download URL: dataforge_sdk-10.3.0rc25-py3-none-any.whl
- Upload date:
- Size: 20.4 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/7.0.0 CPython/3.10.20
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
d2db398b6d2c303d19e6bcfa6c2c55c3a6883d8ccd965c371624e0b4ac531dff
|
|
| MD5 |
c247023fd21cd0b2c0996d3517de78fd
|
|
| BLAKE2b-256 |
0767acbb82e484e9c38a49471708298fe47d7d1b2bfa7d73589b99bf3a2da240
|
