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

pip install mysql-connector-python pandas sqlalchemy psycopg2-binary asyncpg

import mysql.connector
import pandas as pd
from sqlalchemy import create_engine

def create_server_connection(host_name, user_name, user_password):
    connection = None
    try:
        connection = mysql.connector.connect(
            host=host_name,
            user=user_name,
            passwd=user_password
        )
        print("اتصال به سرور MySQL موفق بود")
    except mysql.connector.Error as err:
        print(f"خطا: '{err}'")
    return connection

def create_database(connection, query):
    cursor = connection.cursor()
    try:
        cursor.execute(query)
        print("پایگاه داده با موفقیت ایجاد شد")
    except mysql.connector.Error as err:
        print(f"خطا: '{err}'")

# مثال
create_db_query = "CREATE DATABASE school_db"
connection = create_server_connection("localhost", "root", "your_password")
create_database(connection, create_db_query)

def create_db_connection(host_name, user_name, user_password, db_name):
    connection = None
    try:
        connection = mysql.connector.connect(
            host=host_name,
            user=user_name,
            passwd=user_password,
            database=db_name
        )
        print("اتصال به پایگاه داده موفق بود")
    except mysql.connector.Error as err:
        print(f"خطا: '{err}'")
    return connection

def execute_query(connection, query):
    cursor = connection.cursor()
    try:
        cursor.execute(query)
        connection.commit()
        print("پرس‌وجو با موفقیت اجرا شد")
    except mysql.connector.Error as err:
        print(f"خطا: '{err}'")

create_teacher_table = """
CREATE TABLE teacher (
    teacher_id INT PRIMARY KEY,
    first_name VARCHAR(40) NOT NULL,
    last_name VARCHAR(40) NOT NULL,
    language_1 VARCHAR(3) NOT NULL,
    language_2 VARCHAR(3),
    dob DATE,
    tax_id INT UNIQUE,
    phone_no VARCHAR(20)
);
"""

connection = create_db_connection("localhost", "root", "your_password", "school_db")
execute_query(connection, create_teacher_table)

teacher_data = """
INSERT INTO teacher VALUES
(۱, 'علی', 'احمدی', 'فارسی', 'انگلیسی', '۱۹۸۵-۰۳-۱۵', ۱۲۳۴۵۶۷۸۹۰, '+۹۸۹۱۲۳۴۵۶۷۸۹'),
(۲, 'زهرا', 'محمدی', 'انگلیسی', 'فرانسوی', '۱۹۹۰-۰۷-۲۲', ۹۸۷۶۵۴۳۲۱۰, '+۹۸۹۹۸۷۶۵۴۳۲۱');
"""

execute_query(connection, teacher_data)

def read_query(connection, query):
    cursor = connection.cursor()
    result = None
    try:
        cursor.execute(query)
        result = cursor.fetchall()
        return result
    except mysql.connector.Error as err:
        print(f"خطا: '{err}'")

# اجرای پرس‌وجو
q1 = "SELECT first_name, last_name, phone_no FROM teacher;"
results = read_query(connection, q1)

for result in results:
    print(result)

df = pd.DataFrame(results, columns=['نام', 'نام خانوادگی', 'تلفن'])
print(df)

update_query = """
UPDATE teacher
SET phone_no = '+989001234567'
WHERE teacher_id = 1;
"""
execute_query(connection, update_query)

delete_query = """
DELETE FROM teacher WHERE teacher_id = 2;
"""
execute_query(connection, delete_query)

import asyncio
import asyncpg

async def fetch_teachers():
    conn = await asyncpg.connect(
        user='your_user',
        password='your_password',
        database='school_db',
        host='localhost'
    )
    try:
        rows = await conn.fetch('SELECT * FROM teacher WHERE language_1 = $1', 'فارسی')
        return [dict(row) for row in rows]
    finally:
        await conn.close()

# اجرا
async def main():
    teachers = await fetch_teachers()
    for t in teachers:
        print(t)

asyncio.run(main())

async def create_pool():
    return await asyncpg.create_pool(
        user='user', password='pass', database='school_db', host='localhost',
        min_size=5, max_size=20
    )

async def query_with_pool(pool):
    async with pool.acquire() as conn:
        return await conn.fetch('SELECT * FROM teacher')

# استفاده
async def main():
    pool = await create_pool()
    results = await query_with_pool(pool)
    print(results)
    await pool.close()

asyncio.run(main())

# خطرناک — هرگز این کار را نکنید!
bad_query = f"SELECT * FROM teacher WHERE first_name = '{user_input}'"

# ایمن — همیشه از پارامتر استفاده کنید
safe_query = "SELECT * FROM teacher WHERE first_name = %s"
cursor.execute(safe_query, (user_input,))

connection = mysql.connector.connect(
    host='your-host.com',
    user='user',
    password='pass',
    database='db',
    ssl_ca='/path/to/ca.pem',
    ssl_cert='/path/to/client-cert.pem',
    ssl_key='/path/to/client-key.pem',
    ssl_verify_cert=True
)

from sqlalchemy import create_engine

engine = create_engine(
    'mysql+pymysql://user:pass@localhost/school_db',
    pool_size=10,
    max_overflow=20,
    pool_timeout=30,
    pool_recycle=3600,
    pool_pre_ping=True
)

with engine.connect() as conn:
    result = conn.execute("SELECT * FROM teacher")
    for row in result:
        print(row)

def batch_insert(data):
    query = "INSERT INTO teacher (teacher_id, first_name, last_name) VALUES (%s, %s, %s)"
    cursor = connection.cursor()
    cursor.executemany(query, data)
    connection.commit()
    print(f"{cursor.rowcount} رکورد درج شد")

data = [(3, 'رضا', 'رضایی'), (۴, 'مریم', 'کریمی')]
batch_insert(data)

trigger_sql = """
CREATE TRIGGER before_teacher_update
BEFORE UPDATE ON teacher
FOR EACH ROW
BEGIN
    INSERT INTO teacher_audit (teacher_id, old_name, new_name, changed_at)
    VALUES (OLD.teacher_id, OLD.first_name, NEW.first_name, NOW());
END;
"""
execute_query(connection, trigger_sql)