Implementation:Mage ai Mage ai Source Load Schemas From Folder
| Knowledge Sources | |
|---|---|
| Domains | Data_Integration, Schema_Management |
| Last Updated | 2026-02-09 00:00 GMT |
Overview
Concrete tool for loading JSON Schema files from a connector's schemas directory into Singer Schema objects provided by the Mage integrations Source base class.
Description
Source.load_schemas_from_folder resolves the schemas directory relative to the connector's __init__.py file using inspect.getfile(), lists all .json files, parses each with json.load(), and converts to Singer Schema objects via Schema.from_dict(). The resulting dict maps stream_id (filename without .json) to Schema objects. The schemas_folder name defaults to 'schemas' but can be overridden in __init__.
Usage
Called automatically by Source.discover(). Override in subclasses to customize schema loading (e.g., Chargebee dynamically determines available streams based on product catalog type).
Code Reference
Source Location
- Repository: mage-ai
- File: mage_integrations/mage_integrations/sources/base.py
- Lines: 687-705
Signature
class Source:
def load_schemas_from_folder(self) -> Dict:
"""Load schemas from /schemas folder.
Returns:
Dict mapping stream_id (str) to Schema object.
"""
Import
from mage_integrations.sources.base import Source
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| self.schemas_folder | str | Yes | Directory name (default 'schemas') |
| schemas/*.json | JSON files | Yes | JSON Schema Draft 4 files, one per stream |
Outputs
| Name | Type | Description |
|---|---|---|
| return | Dict[str, Schema] | Maps stream_id to Singer Schema object |
Usage Examples
# Given directory structure:
# sources/my_source/
# __init__.py
# schemas/
# users.json -> {"properties": {"id": {"type": "integer"}, ...}}
# events.json -> {"properties": {"event_id": {"type": "string"}, ...}}
source = MySource()
schemas = source.load_schemas_from_folder()
# schemas = {"users": Schema(...), "events": Schema(...)}