اتصال به دیتابیس در فلسک

اتصال به دیتابیس MySQL/MariaDB

برای اتصال به دیتابیس MySQL ابتدا باید کتابخانه mysql-connector-python را نصب کنیم. برای نصب این کتابخانه از دستور زیر استفاده می‌کنیم:

pip install mysql-connector-python

سپس برای اتصال به دیتابیس MySQL از کد زیر استفاده می‌کنیم:

from flask import Flask, jsonify, render_template
import mysql.connector
import logging
import os

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)

DB_USER = os.environ.get('DB_USER', '')
DB_PASSWORD = os.environ.get('DB_PASSWORD', '')
DB_HOST = os.environ.get('DB_HOST', '')
DB_PORT = os.environ.get('DB_PORT', '')
DB_NAME = os.environ.get('DB_NAME', '')

def test_mysql_connection():
    try:
        conn = mysql.connector.connect(
            host=DB_HOST,
            user=DB_USER,
            password=DB_PASSWORD,
            database=DB_NAME,
            port=DB_PORT
        )
        logging.info("Successfully connected to MySQL")
        conn.close()
        return True
    except Exception as e:
        logging.error(f"MySQL connection failed: {e}")
        return False

@app.route('/')
def home():
    if test_mysql_connection():
        return jsonify({"message": "Successfully connected to MySQL"})
    return jsonify({"message": "Failed to connection to MySQL"})

if __name__ == '__main__':
    app.run(debug=True)

در این کد ابتدا کتابخانه‌های مورد نیاز را import کرده‌ایم. سپس تابع test_mysql_connection را تعریف کرده‌ایم که اتصال به دیتابیس MySQL را بررسی می‌کند. در این تابع ابتدا از متغیرهای محیطی برای تنظیمات دیتابیس استفاده کرده‌ایم. سپس اتصال به دیتابیس را بررسی کرده و در صورت موفقیت به کاربر پیام مناسب برمی‌گرداند.

در تابع home ابتدا از تابع test_mysql_connection برای بررسی اتصال به دیتابیس MySQL استفاده کرده‌ایم. در صورت موفقیت پیام مناسب برگردانده می‌شود.

برای اجرای این برنامه در کوبار باید اپ mysql-client را به run.sh اضافه کنید:

apt-get install default-mysql-client

اتصال به دیتابیس redis

برای اتصال به دیتابیس redis ابتدا باید کتابخانه redis را نصب کنیم. برای نصب این کتابخانه از دستور زیر استفاده می‌کنیم:

pip install redis

سپس برای اتصال به دیتابیس redis از کد زیر استفاده می‌کنیم:

import os
from flask import Flask, request, jsonify
import redis
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = Flask(__name__)

REDIS_PASS = os.environ.get('REDIS_PASS')
REDIS_HOST = os.environ.get('REDIS_HOST')
REDIS_PORT = os.environ.get('REDIS_PORT')

def check_redis_connection():
    try:
        redis_client = redis.Redis(host=REDIS_HOST, port=int(REDIS_PORT), password=REDIS_PASS)
        redis_client.ping()
        logging.info("Successfully connected to Redis")
        return True
    except redis.ConnectionError as e:
        logging.error(f"Failed to connection to Redis: {str(e)}")
        return False

@app.route('/', methods=['GET'])
def home():
    if check_redis_connection():
        return jsonify({"message": "Successfully connected to Redis"})
    return jsonify({"message": "Failed to connection to Redis"})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=int(os.environ.get('PORT', 5000)))

نکته

برای اتصال به دیتابیس‌های دیگر نیز می‌توانید از کتابخانه‌های مربوطه استفاده کنید. برای نصب کتابخانه‌های مربوط به دیتابیس‌های دیگر از دستورات زیر استفاده کنید:

#  postgresql
pip install psycopg2

# mongodb
pip install pymongo

# elasticsearch
pip install elasticsearch