چگونه MySQL را به پایتون متصل کنیم؟

pip install mysql-connector-python

pip install mysql-connector-python==8.4.0

import mysql.connector
print(mysql.connector.__version__)  # خروجی: ۸.۴.۰

import mysql.connector
from mysql.connector import Error, pooling

connection_config = {
    'host': '127.0.0.1',
    'port': 3306,
    'user': 'your_user',
    'password': 'your_strong_password!',
    'database': 'company_db',
    'charset': 'utf8mb4',
    'use_unicode': True,
    'autocommit': False,
    'time_zone': '+00:00',
    'connect_timeout': 10,
    'read_timeout': 30
}

# اتصال SSL برای محیط‌های تولیدی
ssl_config = {
    'ssl_ca': '/etc/ssl/certs/ca-cert.pem',
    'ssl_cert': '/etc/ssl/certs/client-cert.pem',
    'ssl_key': '/etc/ssl/certs/client-key.pem',
    'ssl_verify_cert': True,
    'ssl_verify_identity': True
}

connection_config.update(ssl_config)

try:
    conn = mysql.connector.connect(**connection_config)
    if conn.is_connected():
        cursor = conn.cursor()
        cursor.execute("SELECT VERSION(), DATABASE()")
        version, db = cursor.fetchone()
        print(f"اتصال موفق به MySQL {version} - دیتابیس: {db}")
except Error as e:
    print(f"خطای اتصال: {e}")

from mysql.connector import pooling

db_pool = pooling.MySQLConnectionPool(
    pool_name="production_pool",
    pool_size=15,
    pool_reset_session=True,
    host='mysql-prod.example.com',
    database='analytics',
    user='app_user',
    password='secure_pass',
    ssl_ca='/path/to/ca.pem',
    ssl_verify_cert=True
)

# دریافت اتصال از استخر
conn = db_pool.get_connection()

import aiomysql
import asyncio

async def fetch_data():
    conn = await aiomysql.connect(
        host='localhost', port=3306,
        user='user', password='pass',
        db='company_db', charset='utf8mb4'
    )
    async with conn.cursor() as cur:
        await cur.execute("SELECT * FROM employees LIMIT 10")
        result = await cur.fetchall()
    conn.close()
    return result

# اجرا
data = asyncio.run(fetch_data())
print(data)

def is_connected(conn):
    try:
        conn.ping(reconnect=True, attempts=3, delay=2)
        return True
    except Error:
        return False

def get_valid_connection():
    global conn
    if not is_connected(conn):
        conn = mysql.connector.connect(**connection_config)
    return conn

import time
import logging
from mysql.connector import errors

def execute_query(query, params=None, retries=3):
    for attempt in range(retries):
        try:
            cursor = conn.cursor()
            cursor.execute(query, params)
            result = cursor.fetchall()
            conn.commit()
            return result

        except errors.InterfaceError as e:
            logging.warning(f"اتصال قطع شد (تلاش {attempt+1}): {e}")
            if attempt < retries - 1:
                time.sleep(2 ** attempt)
                global conn
                conn.reconnect()
                continue
            raise

        except errors.OperationalError as e:
            if e.errno == 1213:  # Deadlock
                time.sleep(1)
                continue
            raise

        except (errors.ProgrammingError, errors.DataError) as e:
            logging.error(f"خطای برنامه‌نویسی/داده: {e}")
            raise

        finally:
            cursor.close()

def safe_transaction(operations):
    conn.start_transaction()
    try:
        for query, params in operations:
            cursor = conn.cursor()
            cursor.execute(query, params)
        conn.commit()
        print("تراکنش موفق")
    except Exception as e:
        conn.rollback()
        print(f"تراکنش لغو شد: {e}")
        raise
    finally:
        cursor.close()

class MySQLCircuitBreaker:
    def __init__(self, threshold=5, timeout=60):
        self.threshold = threshold
        self.timeout = timeout
        self.failures = 0
        self.last_failure = None
        self.state = "CLOSED"

    def call(self, func):
        if self.state == "OPEN":
            if time.time() - self.last_failure > self.timeout:
                self.state = "HALF_OPEN"
            else:
                raise Exception("دیتابیس در دسترس نیست")

        try:
            result = func()
            self.failures = 0
            self.state = "CLOSED"
            return result
        except:
            self.failures += 1
            self.last_failure = time.time()
            if self.failures >= self.threshold:
                self.state = "OPEN"
            raise

import airbyte as ab

# تنظیم منبع MySQL
source = ab.get_source(
    "source-mysql",
    config={
        "host": "prod-mysql.example.com",
        "port": 3306,
        "username": "readonly",
        "password": "secure123",
        "database": "sales_db",
        "ssl_mode": "require"
    }
)

# خواندن داده به کش محلی
cache = ab.new_local_cache()
source.read(cache=cache)

# تبدیل به DataFrame
df = cache["orders"].to_pandas()
print(df.head())