Custom source

​

Folder structure

.
├── mage_data
└── project_name
    ├── charts
    ├── custom
    ├── data_exporters
    ├── data_loaders
    ├── dbt
    ├── extensions
    ├── pipelines
    ├── data_integrations              # ← important
    │   └── sources
    │   │   ├── custom_source_1
    │   │   │   └── templates
    │   │   │   │   └── config.json
    │   │   │   └── __init__.py
    │   │   │   └── README.md
    │   │   ├── custom_source_2
    │   │   └── metadata.yaml          # Register all source definitions here
    │   └── destinations
    │       ├── custom_destination_1
    │       ├── custom_destination_2
    │       └── metadata.yaml          # Register all destination definitions here
    ├── scratchpads
    ├── transformers
    ├── utils
    ├── __init__.py
    ├── io_config.yaml
    ├── metadata.yaml
    └── requirements.txt

1. Create a data_integrations folder inside your project directory.
2. Inside data_integrations, create two subfolders:
   • sources: for your custom data source integrations
   • destinations: for your custom data destination integrations
3. Each source/destination should have its own folder (e.g., custom_source_1) containing:
   • An __init__.py file with the integration class
   • A README.md for documentation
   • A templates/ directory with a config.json file for default configuration
4. Define a metadata.yaml file in each of the sources/ and destinations/ folders to register the integrations with Mage Pro.

​

Register custom sources and destinations

sources:
- name: Source Name
  module_name: SourceModuleName
  uuid: source_uuid

destinations:
- name: Source Name
  module_name: SourceModuleName
  uuid: source_uuid

• name: Display name shown in the Mage UI dropdown.
• module_name (optional): The class name defined in the source’s __init__.py file. Defaults to the name field with spaces removed.
• uuid (optional): Folder name of the source. Defaults to the lowercased name, with spaces replaced by underscores.

​

Code modification

# Before
from mage_integrations.sources.custom_source.module import SomeClass

# After
from data_integrations.sources.custom_source.module import SomeClass

​

Testing Your Custom Source

1. Create a new data integration pipeline.
2. Select your custom source from the source dropdown.
   • Custom sources appear at the bottom of the list in the UI.
3. Enter your source configuration.
4. Click “Test connection” if your custom source class implements the test_connection method.
5. Click “View and select stream” to verify that the expected streams are returned.
6. Add a destination and run the pipeline using a trigger to test end-to-end data flow.

​

Troubleshooting: Source Not Appearing in UI?

1. Create a pipeline and add a scratchpad block.
2. Run the following code in the block to inspect the custom sources:

# Inspect custom source registration
from mage_ai.server.api.integration_sources import get_custom_collection

custom_sources = get_custom_collection('sources', debug=True)
print(f'Custom sources detected: {custom_sources}')

3. Try importing your custom module manually to ensure it’s discoverable and error-free:

from data_integrations.sources.source_uuid import SourceModuleName

​

Troubleshooting: “View and select stream” Not Working?

from data_integrations.sources.source_uuid import SourceModuleName

# Replace with your actual config keys and values
config = dict(
    key1='value1',
    key2='value2',
)

# Initialize the source class with config
source = SourceModuleName(config=config)

# Run stream discovery
streams = source.discover_streams()

# Print discovered streams
print(f'Discovered streams: {streams}')

• Ensure the config you provide contains all required fields expected by your source class.
• If discover_streams() returns an empty list or raises an error, check your source implementation or credentials.