نحوه یکسان سازی اطلاعات دیتا بیس های PostgreSQL , MySQL

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta
from sync_module import mysql_to_postgres_sync

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'mysql_postgres_sync',
    default_args=default_args,
    description='Sync data from MySQL to PostgreSQL',
    schedule_interval=timedelta(hours=1),
)

sync_task = PythonOperator(
    task_id='mysql_to_postgres_sync',
    python_callable=mysql_to_postgres_sync,
    dag=dag,
)

-- ایجاد publication در PostgreSQL
CREATE PUBLICATION my_publication FOR TABLE users, orders;

-- ایجاد subscription در PostgreSQL دیگر (برای همگام‌سازی بین دو PostgreSQL)
-- برای MySQL نیاز به یک مکانیزم custom داریم

import mysql.connector
from psycopg2 import connect
import json
from datetime import datetime

class MySQLToPGReplicator:
    def __init__(self, mysql_config, pg_config):
        self.mysql_config = mysql_config
        self.pg_config = pg_config
        self.last_position = None
        
    def connect_to_mysql(self):
        return mysql.connector.connect(**self.mysql_config)
        
    def connect_to_postgres(self):
        return connect(**self.pg_config)
        
    def read_binlog_events(self):
        # پیاده‌سازی خواندن events از MySQL binlog
        pass
        
    def apply_to_postgres(self, events):
        # پیاده‌سازی اعمال تغییرات در PostgreSQL
        pass
        
    def start_replication(self):
        while True:
            events = self.read_binlog_events()
            if events:
                self.apply_to_postgres(events)
                self.save_position()
            time.sleep(0.1)  # تنظیم interval بر اساس نیاز

import pandas as pd
from sqlalchemy import create_engine, text
from datetime import datetime, timedelta
import logging
import sys

# تنظیمات logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('sync_log.log'),
        logging.StreamHandler(sys.stdout)
    ]
)

class DatabaseSynchronizer:
    def __init__(self, mysql_conn_str, postgres_conn_str):
        self.mysql_engine = create_engine(mysql_conn_str)
        self.postgres_engine = create_engine(postgres_conn_str)
        self.sync_state = self.load_sync_state()
        
    def load_sync_state(self):
        """بارگذاری آخرین وضعیت همگام‌سازی"""
        try:
            with self.postgres_engine.connect() as conn:
                result = conn.execute(text("SELECT last_sync_time FROM sync_metadata WHERE table_name = 'target_table'"))
                return result.scalar() or datetime(1970, 1, 1)
        except:
            return datetime(1970, 1, 1)
    
    def save_sync_state(self, new_sync_time):
        """ذخیره زمان آخرین همگام‌سازی"""
        with self.postgres_engine.connect() as conn:
            conn.execute(text("""
                INSERT INTO sync_metadata (table_name, last_sync_time) 
                VALUES ('target_table', :sync_time)
                ON CONFLICT (table_name) 
                DO UPDATE SET last_sync_time = EXCLUDED.last_sync_time
            """), {'sync_time': new_sync_time})
            conn.commit()
    
    def get_changed_data(self, table_name, last_sync):
        """دریافت داده‌های تغییر کرده از MySQL"""
        query = text(f"""
            SELECT * FROM {table_name} 
            WHERE last_modified > :last_sync 
            OR created > :last_sync
        """)
        
        with self.mysql_engine.connect() as conn:
            df = pd.read_sql_query(query, conn, params={'last_sync': last_sync})
        
        return df
    
    def apply_changes_to_postgres(self, df, table_name):
        """اعمال تغییرات در PostgreSQL"""
        if df.empty:
            logging.info("No changes to sync")
            return
        
        # تبدیل نوع داده‌ها اگر لازم باشد
        type_mappings = {
            'tinyint': 'boolean',
            'datetime': 'timestamp'
        }
        
        # ذخیره داده‌ها در PostgreSQL
        with self.postgres_engine.begin() as conn:
            # استفاده از روش UPSERT برای به روزرسانی یا درج رکوردها
            for _, row in df.iterrows():
                # ساختار dynamic query بر اساس ستون‌های موجود
                columns = ', '.join(row.index)
                placeholders = ', '.join([f':{col}' for col in row.index])
                update_clause = ', '.join([f"{col} = EXCLUDED.{col}" for col in row.index if col != 'id'])
                
                upsert_query = text(f"""
                    INSERT INTO {table_name} ({columns}) 
                    VALUES ({placeholders})
                    ON CONFLICT (id) 
                    DO UPDATE SET {update_clause}
                """)
                
                conn.execute(upsert_query, row.to_dict())
        
        logging.info(f"Applied {len(df)} changes to {table_name}")
    
    def sync_table(self, table_name):
        """همگام‌سازی یک جدول"""
        try:
            # دریافت داده‌های تغییر کرده
            changed_data = self.get_changed_data(table_name, self.sync_state)
            
            # اعمال تغییرات
            self.apply_changes_to_postgres(changed_data, table_name)
            
            # به روزرسانی وضعیت همگام‌سازی
            new_sync_time = datetime.now()
            self.save_sync_state(new_sync_time)
            self.sync_state = new_sync_time
            
            logging.info(f"Sync completed for {table_name} at {new_sync_time}")
            
        except Exception as e:
            logging.error(f"Error syncing {table_name}: {str(e)}")
            raise
    
    def sync_all_tables(self):
        """همگام‌سازی تمام جداول تعریف شده"""
        tables = ['users', 'orders', 'products']  # لیست جداول برای همگام‌سازی
        
        for table in tables:
            self.sync_table(table)
    
    def close_connections(self):
        """بستن اتصالات"""
        self.mysql_engine.dispose()
        self.postgres_engine.dispose()

# استفاده از کلاس
if __name__ == "__main__":
    mysql_conn_str = 'mysql+mysqlconnector://user:password@mysql_host/mysql_db'
    postgres_conn_str = 'postgresql+psycopg2://user:password@postgresql_host/postgresql_db'
    
    synchronizer = DatabaseSynchronizer(mysql_conn_str, postgres_conn_str)
    
    try:
        synchronizer.sync_all_tables()
    finally:
        synchronizer.close_connections()

def track_deletes(self):
    """ردیابی رکوردهای حذف شده"""
    # در MySQL، با استفاده از triggers حذف‌ها را در جدول جداگانه ثبت کنید
    pass

def apply_deletes(self):
    """اعمال حذف‌ها در مقصد"""
    with self.mysql_engine.connect() as conn:
        deletes = pd.read_sql_query("SELECT id FROM deletion_log WHERE applied = 0", conn)
        
    if not deletes.empty:
        with self.postgres_engine.begin() as conn:
            for _, row in deletes.iterrows():
                conn.execute(text("DELETE FROM target_table WHERE id = :id"), {'id': row['id']})
            
            # علامت گذاری حذف‌ها به عنوان applied
            conn.execute(text("UPDATE deletion_log SET applied = 1 WHERE id IN :ids"), 
                        {'ids': tuple(deletes['id'].tolist())})

# نصب pgLoader
sudo apt-get install pgloader

# اجرای انتقال
pgloader mysql://user:password@mysql_host/db_name postgresql://user:password@postgresql_host/db_name

def generate_schema_mapping(mysql_schema, postgres_schema):
    """تولید mapping بین طرحواره‌ها"""
    type_mapping = {
        'tinyint(1)': 'boolean',
        'datetime': 'timestamp',
        'longtext': 'text',
        'mediumtext': 'text',
        'varchar(255)': 'varchar(255)',
        'int(11)': 'integer',
        'bigint(20)': 'bigint',
        'double': 'double precision',
        'decimal(10,2)': 'numeric(10,2)'
    }
    
    schema_diff = {}
    
    for table in mysql_schema.tables:
        if table not in postgres_schema.tables:
            schema_diff[table] = {'action': 'create', 'columns': {}}
            continue
            
        for column in mysql_schema.tables[table].columns:
            pg_column = postgres_schema.tables[table].columns.get(column.name)
            
            if not pg_column:
                schema_diff.setdefault(table], {'action': 'alter', 'columns': {}})
                schema_diff[table]['columns'][column.name] = {
                    'action': 'add',
                    'type': type_mapping.get(column.type, column.type)
                }
            elif pg_column.type != type_mapping.get(column.type, column.type):
                schema_diff.setdefault(table], {'action': 'alter', 'columns': {}})
                schema_diff[table]['columns'][column.name] = {
                    'action': 'modify',
                    'old_type': pg_column.type,
                    'new_type': type_mapping.get(column.type, column.type)
                }
    
    return schema_diff

from prometheus_client import Counter, Gauge, start_http_server
import time

# تعریف metrics
SYNC_RECORDS_COUNTER = Counter('sync_records_total', 'Total records synced', ['table'])
SYNC_DURATION_GAUGE = Gauge('sync_duration_seconds', 'Sync duration in seconds')
SYNC_LAG_GAUGE = Gauge('sync_lag_seconds', 'Sync lag in seconds')
SYNC_ERROR_COUNTER = Counter('sync_errors_total', 'Total sync errors')

class MonitoredSynchronizer(DatabaseSynchronizer):
    def sync_table(self, table_name):
        start_time = time.time()
        
        try:
            super().sync_table(table_name)
            duration = time.time() - start_time
            SYNC_DURATION_GAUGE.set(duration)
            SYNC_RECORDS_COUNTER.labels(table=table_name).inc(self.record_count)
            
        except Exception as e:
            SYNC_ERROR_COUNTER.inc()
            logging.error(f"Sync failed for {table_name}: {str(e)}")
            raise

import tenacity
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=4, max=10),
    retry=retry_if_exception_type((NetworkError, DatabaseError))
)
def sync_with_retry(self, table_name):
    """همگام‌سازی با قابلیت retry"""
    return self.sync_table(table_name)

def handle_sync_error(self, table_name, record_id, error):
    """مدیریت خطاهای همگام‌سازی"""
    error_record = {
        'table_name': table_name,
        'record_id': record_id,
        'error_message': str(error),
        'timestamp': datetime.now(),
        'retry_count': 0
    }
    
    # ذخیره در جدول error queue
    with self.postgres_engine.begin() as conn:
        conn.execute(
            text("INSERT INTO sync_error_queue (table_name, record_id, error_message, timestamp, retry_count) VALUES (:table_name, :record_id, :error_message, :timestamp, :retry_count)"),
            error_record
        )