Custom Connectors

To connect with sources and destinations that are not currently supported natively in Tabsdata, you can use these connector plugins.

Source Plugin

The SourcePlugin class enables reading from sources which do not have built-in support in Tabsdata. Click a numbered marker (or the legend beside it) to see what each part does.

source_plugin.py

1import os
2import polars as pl
3import requests
4import tabsdata as td
5 
1class PyPIPkgStatsSource(td.SourcePlugin):
7    def __init__(self, package_name: str):
8        self.package_name = package_name
9 
2    def chunk(self, working_dir: str) -> str:
11        # Endpoint with the downloads information of the package
12        base_endpoint = f"https://pypistats.org/api/packages/{self.package_name}"
13        # Get the downloads by system
14        downloads_by_system = requests.get(f"{base_endpoint}/system").json().get("data")
15 
16        # Store the information
17        destination_file = "data.parquet"
18        destination_path = os.path.join(working_dir, destination_file)
19        pl.DataFrame(downloads_by_system).write_parquet(destination_path)
3        return destination_file
21 
22@td.publisher(
23    source=PyPIPkgStatsSource("polars"),
24    tables="output",
25)
26def input_plugin_from_pypi(df: td.TableFrame):
27    return df

• Inherit from td.SourcePlugin and accept whatever configuration your source needs, which here is a PyPI package name.

• The one method you must implement. Fetch or generate the data, write it as a file inside working_dir, and return that file's name. Tabsdata reads the file you return.

• Pass an instance of your plugin as the publisher's source, exactly like a built-in connector.

Destination Plugin

The DestinationPlugin class enables writing to destinations which do not have built-in support in Tabsdata.

destination_plugin.py

1import os
2import tempfile
3import polars as pl
4from google.cloud import storage
5import tabsdata as td
6 
1class GCPFileUpload(td.DestinationPlugin):
8    def __init__(self, bucket_name: str, gcp_credentials_path: str):
9        self.bucket_name = bucket_name
10        self.gcp_credentials_path = gcp_credentials_path
2 
12    def stream(self, _: str, lf: pl.LazyFrame):
13        # Set the GCP credentials path
14        os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = self.gcp_credentials_path
15 
16        # Create a temporary file
17        with tempfile.NamedTemporaryFile(suffix=".csv") as tmp_file:
18            temp_file_path = tmp_file.name
19            lf.sink_csv(temp_file_path)
20            # Extract filename
21            destination_file = os.path.basename(temp_file_path)
22            # Upload the file to GCP Storage
23            client = storage.Client()
24            bucket = client.bucket(self.bucket_name)
25            blob = bucket.blob(destination_file)
3            blob.upload_from_filename(temp_file_path)
27 
28@td.subscriber(
29    tables="data",
30    destination=GCPFileUpload("<gcp-bucket-name>", "<path_to_gcp_credentials.json>"),
31)
32def subscriber(df: pl.DataFrame):
33    return df

• Inherit from td.DestinationPlugin and accept whatever configuration your destination needs, which here is a GCP bucket name and a credentials path.

• The one method you must implement. Receives the table as a Polars LazyFrame and is responsible for getting it to the external system. Here that means writing a temp CSV and uploading it to GCS.

• Pass an instance of your plugin as the subscriber's destination, exactly like a built-in connector.

Where:

• <gcp-bucket-name> is the name of your GCP bucket.
• <path_to_gcp_credentials.json> is the full system path (typically starting with /users/user_name) to your gcp credentials file.