ObjDataTransfer API

Method	Signature	Description
load_yaml_query	`load_yaml_query(query_name: str) -> str`	Load a SQL query from the ObjDataTransfer.yaml file.
update_progress	`update_progress(track_guid: str, records_processed: int, total_records: int, phase: str = 'WRITING', start_time: datetime = None, bytes_processed: int = 0) -> None`	Update real-time progress tracking for active transfer.
validate_transfer	`validate_transfer(track_guid: str, transfer_code: str, source_table: str, target_table: str, filter_query: str = '1=1', filter_guid: str = 'guid', validation_types: list = None) -> dict`	Validate transfer results.
transfer_incremental	`transfer_incremental(transfer_code: str, dry_run: bool = False) -> int`	Execute incremental transfer (only changed records since last run).
transfer_with_retry	`transfer_with_retry(transfer_code: str, max_retries: int = None, backoff_multiplier: float = 2.0) -> int`	Execute transfer with retry logic and exponential backoff.
transfer_dry_run	`transfer_dry_run(transfer_code: str, sample_size: int = 10) -> dict`	Simulate transfer without writing data.
initialize_notification_codes	`initialize_notification_codes() -> bool`	Initialize notification codes for data transfer tracking.
get_connection_type	`get_connection_type(remote: str) -> str`	Detect connection type from remote connection or file path.
track_transfer_start	`track_transfer_start(transfer_code: str, source_table: str, target_table: str, transfer_mode: str = TransferMode.SUPPLEMENT.value) -> str`	Record the start of a transfer in data_track_bulktransfer.
track_transfer_end	`track_transfer_end(track_guid: str, status: str, records_transferred: int = 0, records_deleted: int = 0, error_message: str = '', error_details: str = '') -> None`	Record the completion of a transfer in data_track_bulktransfer.
remote_connect_source	`remote_connect_source(remote: str) -> None`
remote_connect_target	`remote_connect_target(remote: str) -> None`
is_type_compatible	`is_type_compatible(source_type: str, target_type: str) -> bool`	Check if source_type from Mongo is compatible with target_type in SQL.
yaml_read_file	`yaml_read_file(file_path: str) -> list[dict[str, Any]]`	Read YAML file and return list of records.
yaml_write_file	`yaml_write_file(file_path: str, data: list[dict[str, Any]], wrapper_key: str = 'records') -> bool`	Write data to YAML file.
yaml_infer_schema	`yaml_infer_schema(data: list[dict[str, Any]]) -> list[dict[str, str]]`	Infer schema from YAML data.
yaml_get_fields_from_file	`yaml_get_fields_from_file(file_path: str) -> list[dict[str, str]]`	Get field schema from YAML file.
transfer_query	`transfer_query(source_connection: str, source_query: str, target_connection: str, target_table: str, primary_keys: list[str], batch_size: int = 500, create_target: bool = True) -> int`	Ad-hoc transfer: execute a source query and upsert results into a target.
validate_cron_schedule	`validate_cron_schedule(cron_expression: str) -> bool`	Validate a cron expression.
should_run_transfer	`should_run_transfer(cron_expression: str, last_transfer: datetime \| None = None) -> bool`	Check if a transfer should run based on its cron schedule.
validate_structure	`validate_structure(table_source: str, table_target: str) -> bool`
transfer_structure	`transfer_structure(table_source: str, table_target: str, filter_guid: str, filter_query: str, buffer_size: int = 500, transfer_mode: str = TransferMode.SUPPLEMENT.value, transfer_code: str = '', enable_tracking: bool = True) -> int`
transfer_bulk_set	`transfer_bulk_set() -> None`
transfer	`transfer(transfercode: str) -> int`
run_workflow_direct	`run_workflow_direct(Guid: str, Param1: str = '', Param2: str = '', Param3: str = '') -> str`

¶ ObjDataTransfer (API Reference)

¶ class ObjDataTransfer(ObjData.ObjData)

¶ transferbulk()

¶ transfer(transfer_code: str)