Implementation:Mage ai Mage ai Parse Args
| Knowledge Sources | |
|---|---|
| Domains | Data_Integration, ETL |
| Last Updated | 2026-02-09 00:00 GMT |
Overview
Concrete tool for parsing Singer tap CLI arguments and initializing source configuration provided by the Mage integrations framework.
Description
The parse_args function handles the complete CLI argument parsing for Singer-compatible source connectors. It uses Python's argparse to define standard Singer flags (-c/--config, -s/--state, --catalog, -d/--discover) plus Mage-specific extensions (--config_json, --state_json, --catalog_json, --query_json, --selected_streams_json, --settings). After parsing, it loads JSON files, resolves YAML settings, constructs a Catalog object, and validates required config keys.
Usage
Import this function when building any source connector that needs to parse CLI arguments. Called internally by Source.__init__ but can be used standalone for custom entry points.
Code Reference
Source Location
- Repository: mage-ai
- File: mage_integrations/mage_integrations/sources/utils.py
- Lines: 145-318
Signature
def parse_args(required_config_keys: List[str]) -> argparse.Namespace:
"""Parse standard command-line args.
Args:
required_config_keys: List of config keys that must be present.
Returns:
argparse.Namespace with attributes:
config (Dict), state (Dict), catalog (Catalog),
query (Dict), selected_streams (List[str]),
discover (bool), count_records (bool),
test_connection (bool), load_sample_data (bool),
show_templates (bool), log_to_stdout (bool)
"""
Import
from mage_integrations.sources.utils import parse_args
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| required_config_keys | List[str] | Yes | Config keys that must be present in the loaded config |
| CLI: -c/--config | str | No | Path to JSON config file |
| CLI: --config_json | str | No | Inline JSON string with config values |
| CLI: -s/--state | str | No | Path to JSON state file |
| CLI: --state_json | str | No | Inline JSON string with state bookmarks |
| CLI: --catalog | str | No | Path to catalog file |
| CLI: --catalog_json | str | No | Inline JSON string with catalog streams |
| CLI: --settings | str | No | Path to YAML settings file bundling config and catalog |
| CLI: -d/--discover | flag | No | Enable schema discovery mode |
| CLI: --selected_streams_json | str | No | JSON list of stream names to select |
Outputs
| Name | Type | Description |
|---|---|---|
| args.config | Dict | Merged config from file, JSON string, and settings |
| args.state | Dict | Parsed state dict (empty dict if not provided) |
| args.catalog | Catalog | Catalog object from file, JSON, or settings |
| args.query | Dict | Query parameters for load_data filtering |
| args.selected_streams | List[str] | Stream names to include in sync |
| args.discover | bool | Whether to run in discover mode |
| args.test_connection | bool | Whether to test connection only |
Usage Examples
Standard Entry Point
from mage_integrations.sources.utils import parse_args
# Parse CLI args with required database config keys
args = parse_args(required_config_keys=["host", "port", "database", "username", "password"])
# Access parsed config
print(args.config) # {"host": "localhost", "port": 5432, ...}
print(args.state) # {"bookmarks": {"users": {"updated_at": "2024-01-01"}}}
print(args.discover) # True/False based on -d flag
Used Inside Source.__init__
from mage_integrations.sources.base import Source
class MySource(Source):
def __init__(self, **kwargs):
super().__init__(**kwargs)
# parse_args is called automatically in Source.__init__
# self.config, self.state, self.catalog are now populated