Implementation:Mage ai Mage ai GitHub Sync
| Knowledge Sources | |
|---|---|
| Domains | Data_Integration, GitHub, Sync_Orchestration |
| Last Updated | 2026-02-09 00:00 GMT |
Overview
Orchestrates the sync lifecycle for the Mage GitHub source connector, coordinating stream selection, repository ordering, state management, and schema emission.
Description
This module contains the sync orchestration logic for the GitHub source connector. It determines which streams are selected in the catalog, resolves parent-child dependencies so that parent streams are synced even if only a child is selected, manages the currently_syncing and currently_syncing_repo state properties for resumable syncs, translates legacy single-repository state formats into the multi-repository format, orders repositories and streams for deterministic resumption, and writes schemas before syncing data. Key functions include get_selected_streams, update_currently_syncing, update_currently_syncing_repo, get_ordered_stream_list, get_ordered_repos, translate_state, get_stream_to_sync, is_any_child_selected, and write_schemas.
Usage
Called from the top-level tap entry point to execute a full sync. The sync function iterates over repositories and streams, delegates record extraction to the stream classes and client, and maintains Singer state throughout.
Code Reference
Source Location
- Repository: mage-ai
- File: mage_integrations/mage_integrations/sources/github/tap_github/sync.py
- Lines: 1-306
Signature
def get_selected_streams(catalog: dict) -> list:
...
def update_currently_syncing(state: dict, stream_name: str) -> None:
...
def update_currently_syncing_repo(state: dict, repo_path: str) -> None:
...
def get_ordered_stream_list(currently_syncing: str, streams_to_sync: list) -> list:
...
def get_ordered_repos(state: dict, repositories: list) -> list:
...
def translate_state(state: dict, catalog: dict, repositories: list) -> dict:
...
def get_stream_to_sync(catalog: dict) -> list:
...
def is_any_child_selected(stream_obj, selected_streams: list) -> bool:
...
def write_schemas(stream_id: str, catalog: dict, selected_streams: list) -> None:
...
Import
from mage_integrations.sources.github.tap_github.sync import (
get_selected_streams,
get_stream_to_sync,
translate_state,
)
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| catalog | dict | Yes | Singer catalog with stream schemas and selected metadata |
| state | dict | Yes | Singer state with bookmark values and currently_syncing markers |
| repositories | list[str] | Yes | List of repository paths in owner/repo format |
Outputs
| Name | Type | Description |
|---|---|---|
| selected_streams | list[str] | List of selected stream tap_stream_ids |
| streams_to_sync | list[str] | Ordered list of streams to sync including parent dependencies |
| state | dict | Updated Singer state with translated bookmarks and sync markers |
Key Behaviors
State Translation
The translate_state function migrates older single-repository state format (where bookmarks are keyed by stream name) to the multi-repository format (where bookmarks are keyed by repo_path/stream_name). It returns the existing state unchanged if it already uses the new format.
Resumable Syncing
update_currently_syncing and update_currently_syncing_repo track which stream and repository are actively syncing. On restart, get_ordered_stream_list and get_ordered_repos reorder the sync queue so that the interrupted stream/repo is processed first, followed by the remaining items.
Child Stream Resolution
get_stream_to_sync includes parent streams in the sync list even when only a child is selected, because parent data is required to construct child API URLs. is_any_child_selected recursively checks nested children.
Usage Examples
from mage_integrations.sources.github.tap_github.sync import (
get_selected_streams,
get_stream_to_sync,
translate_state,
)
catalog = {"streams": [...]}
state = {"bookmarks": {"commits": {"since": "2024-01-01T00:00:00Z"}}}
repositories = ["mage-ai/mage-ai"]
# Translate legacy state
new_state = translate_state(state, catalog, repositories)
# Get streams to sync
selected = get_selected_streams(catalog)
streams = get_stream_to_sync(catalog)